feat(plugins): add minimal original-user runner integration (#9017)

* feat(plugins): add minimal original-user runner integration

* test(plugins): use the main storage fixture

* fix(plugins): reuse runner connections across requests

* test(plugins): allocate stack for delegation fixture

.config/nextest.toml

@@ -128,7 +128,7 @@ setup = 'ecstore-large-stack'
 # path on a real store; measured with RUST_MIN_STACK on 2026-10-15 they need
 # 2016-2040 KiB of the 2 MiB libtest default and overflowed on Linux CI.
 [[profile.default.scripts]]
-filter = '(package(rustfs) & test(/^(admin::handlers::replication::target_repair_tests::[a-z_]+|admin::handlers::site_replication::tests::durable_iam_state_apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete|admin::service::federated_identity::tests::(binding_keeps_explicit_iam_mapping_strict|markerless_legacy_oidc_session_crosses_auth_and_iam_deletion_boundaries)|storage::scanner_dirty_journal::tests::(clear_callback_keeps_unknown_coverage_until_all_pending_work_is_confirmed|owned_journals_keep_peer_records_after_interleaved_writes_and_clear))$/)) | (package(rustfs-heal) & ((binary(heal_b5_versioned_regression_test) & test(/^serial_tests::(test_heal_unversioned_bucket_e2e|test_historical_null_recursive_heal_restores_physical_versions|test_recursive_deep_heal_reports_stale_delete_marker_repair)$/)) | (binary(usage_observation_read_repair_test) & test(=usage_observation_read_repair_survives_snapshot_cleanup)))) | (package(rustfs-test-utils) & test(=data_usage_snapshot_tests::data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache))'
+filter = '(package(rustfs) & test(/^(admin::handlers::replication::target_repair_tests::[a-z_]+|admin::handlers::plugin_attempt::v2_tests::delegation_v2_persists_and_vends_with_real_parent_intersection|admin::handlers::site_replication::tests::durable_iam_state_apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete|admin::service::federated_identity::tests::(binding_keeps_explicit_iam_mapping_strict|markerless_legacy_oidc_session_crosses_auth_and_iam_deletion_boundaries)|storage::scanner_dirty_journal::tests::(clear_callback_keeps_unknown_coverage_until_all_pending_work_is_confirmed|owned_journals_keep_peer_records_after_interleaved_writes_and_clear))$/)) | (package(rustfs-heal) & ((binary(heal_b5_versioned_regression_test) & test(/^serial_tests::(test_heal_unversioned_bucket_e2e|test_historical_null_recursive_heal_restores_physical_versions|test_recursive_deep_heal_reports_stale_delete_marker_repair)$/)) | (binary(usage_observation_read_repair_test) & test(=usage_observation_read_repair_survives_snapshot_cleanup)))) | (package(rustfs-test-utils) & test(=data_usage_snapshot_tests::data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache))'
 setup = 'ecstore-base-stack'
 
 [[profile.default.scripts]]
@@ -419,7 +419,7 @@ setup = 'ecstore-large-stack'
 
 # These cross-crate real-storage fixtures need the same bounded stack as ECStore tests.
 [[profile.ci.scripts]]
-filter = '(package(rustfs) & test(/^(admin::handlers::replication::target_repair_tests::[a-z_]+|admin::handlers::site_replication::tests::durable_iam_state_apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete|admin::service::federated_identity::tests::(binding_keeps_explicit_iam_mapping_strict|markerless_legacy_oidc_session_crosses_auth_and_iam_deletion_boundaries)|storage::scanner_dirty_journal::tests::(clear_callback_keeps_unknown_coverage_until_all_pending_work_is_confirmed|owned_journals_keep_peer_records_after_interleaved_writes_and_clear))$/)) | (package(rustfs-heal) & ((binary(heal_b5_versioned_regression_test) & test(/^serial_tests::(test_heal_unversioned_bucket_e2e|test_historical_null_recursive_heal_restores_physical_versions|test_recursive_deep_heal_reports_stale_delete_marker_repair)$/)) | (binary(usage_observation_read_repair_test) & test(=usage_observation_read_repair_survives_snapshot_cleanup)))) | (package(rustfs-test-utils) & test(=data_usage_snapshot_tests::data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache))'
+filter = '(package(rustfs) & test(/^(admin::handlers::replication::target_repair_tests::[a-z_]+|admin::handlers::plugin_attempt::v2_tests::delegation_v2_persists_and_vends_with_real_parent_intersection|admin::handlers::site_replication::tests::durable_iam_state_apply_iam_item_rejects_a_stale_recreate_after_a_replicated_delete|admin::service::federated_identity::tests::(binding_keeps_explicit_iam_mapping_strict|markerless_legacy_oidc_session_crosses_auth_and_iam_deletion_boundaries)|storage::scanner_dirty_journal::tests::(clear_callback_keeps_unknown_coverage_until_all_pending_work_is_confirmed|owned_journals_keep_peer_records_after_interleaved_writes_and_clear))$/)) | (package(rustfs-heal) & ((binary(heal_b5_versioned_regression_test) & test(/^serial_tests::(test_heal_unversioned_bucket_e2e|test_historical_null_recursive_heal_restores_physical_versions|test_recursive_deep_heal_reports_stale_delete_marker_repair)$/)) | (binary(usage_observation_read_repair_test) & test(=usage_observation_read_repair_survives_snapshot_cleanup)))) | (package(rustfs-test-utils) & test(=data_usage_snapshot_tests::data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache))'
 setup = 'ecstore-base-stack'
 
 [[profile.ci.scripts]]

.github/workflows/README.md

@@ -16,7 +16,7 @@ Every other label in this repository is capacity RustFS pays for. ARC pods and l
 | `sm-standard-2` | RustFS | ARC pod, Ubuntu 24.04, minimal image: no `gh`, no C toolchain; Docker only on the `dind-` label | only a job that a hosted runner cannot run (see the placement rule) |
 | `sm-standard-4` | RustFS | ARC pod, Ubuntu 24.04, 14 CPUs and 28 GB (the label is not the core count), large disk | workspace-scale Cargo builds |
 | `ubicloud-standard-16-arm` | RustFS | Ubicloud VM, alias of `ubicloud-standard-16-arm-ubuntu-2404`: Ubuntu 24.04 arm64, 16 vCPU, 48 GB, 300 GB disk. Reached through the `build.yml` matrix, so `actionlint.yaml` does not list it | native aarch64 release builds (`linux-aarch64-musl`, `linux-aarch64-gnu`) |
-| `dind-sm-standard-2` | RustFS | ARC pod with Docker-in-Docker | image builds that need `buildx` and a registry push |
+| `dind-sm-standard-2` | RustFS | ARC pod with Docker-in-Docker | Docker workloads with a measured resource or environment requirement beyond the hosted runner |
 | `smoke-testing`, `pf-testing` | RustFS | long-lived lab machines | the packaged functional chain and performance suites |
 
 The self-hosted pods and the hosted image are both Ubuntu 24.04, so a binary built on one runs on the other; `ci.yml` builds the debug binary on `sm-standard-4` and runs it on both pools.
@@ -27,11 +27,15 @@ A job runs on `ubuntu-latest` unless it has one of these reasons, and the reason
 
 1. **Disk.** It restores or produces the workspace `target/`: `cargo build` or `cargo test` with `--all-targets`, `cargo nextest run --all`, or any reader of the `ci-dev` / `ci-feat-*` caches. The workspace test build leaves about 64 GB in `target/`; a hosted runner has about 14 GB.
 2. **Critical-path CPU.** It is a build that several downstream lanes wait on and would blow its timeout at 4 vCPU (`build-rustfs-debug-binary` gates the e2e and s3-tests lanes).
-3. **Hardware.** Native aarch64, Docker-in-Docker, or the lab machines.
+3. **Hardware or environment.** Native aarch64 or access to the lab machines. Docker alone is not a reason: hosted Ubuntu ships Docker and supports the QEMU and Buildx setup actions.
 4. **Cache ownership.** It reads a `rust-cache` key that a self-hosted job writes (see the cache rule below).
 
 Jobs that do only the following have no reason to be self-hosted: shell or Python scripts, `cargo fmt`, `cargo deny`, typos, actionlint, API calls (`gh`, dependency review, issue and stale bots, CLA checks), downloading an artifact and running a prebuilt binary, or a `cargo build -p rustfs` with its own cache key. `e2e-distributed.yml` and `e2e-upgrade.yml` build the server and the `e2e_test` harness on `ubuntu-latest` inside their budgets, so "it compiles something" alone is not a reason.
 
+The functional chain's `prepare` and `complete-chain` jobs, the performance runner availability check, and runner hygiene use `ubuntu-latest`: they inspect APIs or artifacts, not the lab hosts. The actual functional and performance suites remain on their lab runners. Functional-chain health collection stays on the operator-managed `smoke-testing` runner as part of the automation environment. Docker image packaging also uses `ubuntu-latest` for both musl and glibc variants; it downloads release binaries rather than compiling the workspace, and still sets up QEMU and Buildx explicitly.
+
+Moving these jobs off AKS reduces capacity demand, not necessarily the Azure bill immediately. VM costs fall only when nodes scale down or smaller capacity replaces them.
+
 A hosted runner is also the only choice for a few things the pods cannot do: `mount -o loop` and `mount -t tmpfs` fail inside the ARC pods, and a container runtime cannot guarantee that `io_uring_setup` is allowed. `uring-integration` and `e2e-distributed` are on `ubuntu-latest` for that reason, not for cost.
 
 ## Cache rule

.github/workflows/docker.yml

@@ -337,7 +337,7 @@ jobs:
     name: Build Docker Images
     needs: build-check
     if: needs.build-check.outputs.should_build == 'true'
-    runs-on: dind-sm-standard-2
+    runs-on: ubuntu-latest
     timeout-minutes: 60
     strategy:
       fail-fast: false

.github/workflows/runner-hygiene.yml

@@ -44,7 +44,7 @@ jobs:
   check-ephemerality:
     if: github.repository == 'rustfs/rustfs'
     name: Check runner ephemerality
-    runs-on: sm-standard-2
+    runs-on: ubuntu-latest
     timeout-minutes: 15
     steps:
       - name: Checkout repository

.github/workflows/rustfs-functional-chain.yml

@@ -42,7 +42,7 @@ concurrency:
 jobs:
   prepare:
     if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.event == 'schedule' }}
-    runs-on: smoke-testing
+    runs-on: ubuntu-latest
     timeout-minutes: 10
     outputs:
       manifest: ${{ steps.candidate.outputs.manifest }}
@@ -188,7 +188,7 @@ jobs:
   complete-chain:
     needs: [prepare, upgrade, s3, kms, tier, storage, heal, pool, security, replication, fault-tolerance, table, performance]
     if: ${{ always() && needs.prepare.result != 'skipped' }}
-    runs-on: smoke-testing
+    runs-on: ubuntu-latest
     timeout-minutes: 10
     steps:
       - uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7

.github/workflows/rustfs-performance-test.yml

@@ -90,7 +90,7 @@ env:
 jobs:
   runner-check:
     if: ${{ inputs.chain_manifest != '' || github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
-    runs-on: sm-standard-2
+    runs-on: ubuntu-latest
     timeout-minutes: 5
     steps:
       - uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7

crates/concurrency/Cargo.toml

@@ -26,7 +26,7 @@ rustfs-io-core = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 
 # Async runtime
-tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread"] }
+tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread", "time"] }
 
 # Logging
 tracing = { workspace = true }

crates/ecstore/src/api/mod.rs

@@ -371,6 +371,15 @@ pub mod compression {
 
 pub mod config {
     pub mod com {
+        /// Preserve empty existing objects so callers can distinguish corruption from absence.
+        pub async fn read_config_preserve_empty_with_metadata_opts(
+            api: std::sync::Arc<crate::store::ECStore>,
+            file: &str,
+            opts: &crate::object_api::ObjectOptions,
+        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
+            crate::config::com::read_config_preserve_empty_with_metadata_opts(api, file, opts).await
+        }
+
         pub use crate::config::com::{
             COMMA_SEPARATED_LISTS, CONFIG_PREFIX, ENV_CONFIG_RECOVER_ON_CORRUPTION, STORAGE_CLASS_SUB_SYS,
             ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, delete_config_no_lock,

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -1086,8 +1086,9 @@ impl ExpiryState {
         }))
     }
 
-    #[cfg(test)]
-    fn new_with_unconsumed_worker_channel(capacity: usize) -> Arc<RwLock<Self>> {
+    #[cfg(any(test, feature = "test-util"))]
+    #[doc(hidden)]
+    pub fn new_with_unconsumed_worker_channel(capacity: usize) -> Arc<RwLock<Self>> {
         let (tx, rx) = mpsc::channel(capacity);
         Arc::new(RwLock::new(Self {
             tasks_tx: vec![tx],
@@ -3516,13 +3517,21 @@ async fn stale_upload_current_size_with_opts(
         .await
         .ok()?;
 
-    Some(
-        parts
-            .parts
-            .iter()
-            .map(|part| part.actual_size.max(part.size as i64).max(0) as usize)
-            .sum(),
-    )
+    stale_upload_size_from_parts(&parts)
+}
+
+fn stale_upload_size_from_parts(parts: &crate::storage_api_contracts::multipart::ListPartsInfo) -> Option<usize> {
+    if parts.is_truncated {
+        return None;
+    }
+    parts.parts.iter().try_fold(0usize, |total, part| {
+        // ListParts represents a quorum-confirmed missing part with an empty ETag.
+        if part.etag.as_deref().is_none_or(str::is_empty) {
+            return None;
+        }
+        let actual_size = usize::try_from(part.actual_size.max(0)).ok()?;
+        total.checked_add(part.size.max(actual_size))
+    })
 }
 
 async fn stale_upload_lifecycle_due(
@@ -3539,18 +3548,43 @@ async fn stale_upload_lifecycle_due(
     let bucket = metadata.get(RUSTFS_MULTIPART_BUCKET_KEY)?;
     let object = metadata.get(RUSTFS_MULTIPART_OBJECT_KEY)?;
 
-    let lifecycle = match metadata_boundary::get_lifecycle_config(bucket).await {
+    let mut lifecycle = match metadata_boundary::get_lifecycle_config(bucket).await {
         Ok((lifecycle, _)) => lifecycle,
         Err(_) => return None,
     };
 
+    let has_size_filter = |filter: &s3s::dto::LifecycleRuleFilter| {
+        filter.object_size_greater_than.is_some()
+            || filter.object_size_less_than.is_some()
+            || filter
+                .and
+                .as_ref()
+                .is_some_and(|and| and.object_size_greater_than.is_some() || and.object_size_less_than.is_some())
+    };
+    let needs_size = lifecycle
+        .rules
+        .iter()
+        .any(|rule| rule.abort_incomplete_multipart_upload.is_some() && rule.filter.as_ref().is_some_and(has_size_filter));
+    let size = if needs_size {
+        match stale_upload_current_size_with_opts(set, metadata, upload_dir, no_lock).await {
+            Some(size) => size,
+            None => {
+                // An unknown size cannot authorize a size-filtered abort.
+                // Unrelated rules can still be evaluated independently.
+                lifecycle
+                    .rules
+                    .retain(|rule| !rule.filter.as_ref().is_some_and(has_size_filter));
+                0
+            }
+        }
+    } else {
+        0
+    };
     let object_opts = ObjectOpts {
         name: object.clone(),
         user_tags: metadata.get(AMZ_OBJECT_TAGGING).cloned().unwrap_or_default(),
         mod_time: Some(initiated),
-        size: stale_upload_current_size_with_opts(set, metadata, upload_dir, no_lock)
-            .await
-            .unwrap_or_default(),
+        size,
         is_latest: true,
         delete_marker: false,
         user_defined: metadata.clone(),
@@ -3803,7 +3837,8 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc<SetDisks>, now: Offset
     for candidate in candidates.into_values() {
         let upload_dir = candidate.path.rsplit('/').next().unwrap_or_default().to_string();
         let mut due = stale_upload_default_due(candidate.initiated, default_expiry);
-        if let Some(metadata) = candidate.metadata.as_ref()
+        if now < due
+            && let Some(metadata) = candidate.metadata.as_ref()
             && let Some(lifecycle_due) = stale_upload_lifecycle_due(set, metadata, candidate.initiated, &upload_dir, false).await
             && lifecycle_due < due
         {
@@ -3832,8 +3867,9 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc<SetDisks>, now: Offset
         let current_metadata = cleanup_guard.file_info().metadata.clone();
         let current_initiated = initiated_from_upload_dir(&upload_dir, cleanup_guard.file_info().mod_time);
         let mut current_due = stale_upload_default_due(current_initiated, default_expiry);
-        if let Some(lifecycle_due) =
-            stale_upload_lifecycle_due(set, &current_metadata, current_initiated, &upload_dir, true).await
+        if now < current_due
+            && let Some(lifecycle_due) =
+                stale_upload_lifecycle_due(set, &current_metadata, current_initiated, &upload_dir, true).await
             && lifecycle_due < current_due
         {
             current_due = lifecycle_due;
@@ -15868,6 +15904,261 @@ mod tests {
         assert!(ecstore.is_ok(), "fresh local ECStore boot should succeed, got {ecstore:?}");
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    #[serial]
+    async fn stale_multipart_cleanup_preserves_partial_metadata_without_upload_part_warning() {
+        use tracing::instrument::WithSubscriber as _;
+        #[derive(Clone, Default)]
+        struct CapturedLogs(Arc<StdMutex<Vec<u8>>>);
+        struct LogWriter(Arc<StdMutex<Vec<u8>>>);
+        impl std::io::Write for LogWriter {
+            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+                self.0.lock().unwrap().extend_from_slice(bytes);
+                Ok(bytes.len())
+            }
+            fn flush(&mut self) -> std::io::Result<()> {
+                Ok(())
+            }
+        }
+        impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs {
+            type Writer = LogWriter;
+            fn make_writer(&'a self) -> Self::Writer {
+                LogWriter(self.0.clone())
+            }
+        }
+
+        let (paths, ecstore) = setup_test_env().await;
+        let bucket = format!("stale-partial-{}", Uuid::new_v4().simple());
+        create_test_bucket(&ecstore, &bucket).await;
+        for copies in [1, 2] {
+            let object = format!("partial-{copies}.bin");
+            let initiated = OffsetDateTime::now_utc() - time::Duration::hours(30);
+            let upload = ecstore
+                .new_multipart_upload(
+                    &bucket,
+                    &object,
+                    &ObjectOptions {
+                        mod_time: Some(initiated),
+                        max_parity: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("multipart upload should be created");
+            let decoded = base64_simd::URL_SAFE_NO_PAD
+                .decode_to_vec(upload.upload_id.as_bytes())
+                .unwrap();
+            let decoded = String::from_utf8(decoded).unwrap();
+            let (_, upload_dir) = decoded.split_once('.').expect("deployment-prefixed upload id");
+            let upload_path = format!("{}/{upload_dir}", multipart_sha_dir(&bucket, &object));
+            let physical_paths: Vec<_> = paths
+                .iter()
+                .map(|path| path.join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_path))
+                .collect();
+            let original: Vec<_> = physical_paths
+                .iter()
+                .take(copies)
+                .map(|path| std::fs::read(path.join(STORAGE_FORMAT_FILE)).unwrap())
+                .collect();
+            let info = super::get_file_info(
+                &original[0],
+                RUSTFS_META_MULTIPART_BUCKET,
+                &upload_path,
+                "",
+                super::FileInfoOpts {
+                    data: false,
+                    include_free_versions: false,
+                    include_part_checksums: false,
+                },
+            )
+            .unwrap();
+            assert_eq!((info.erasure.data_blocks, info.erasure.parity_blocks), (2, 2));
+            for path in physical_paths.iter().skip(copies) {
+                fs::remove_dir_all(path).await.expect("remove selected metadata copies");
+            }
+            if copies == 2 {
+                ecstore
+                    .get_multipart_info(&bucket, &object, &upload.upload_id, &ObjectOptions::default())
+                    .await
+                    .expect("two metadata copies must meet read quorum");
+            }
+
+            let logs = CapturedLogs::default();
+            let subscriber = tracing_subscriber::fmt()
+                .with_max_level(tracing::Level::DEBUG)
+                .with_writer(logs.clone())
+                .with_ansi(false)
+                .without_time()
+                .finish();
+            cleanup_stale_multipart_uploads_once_at(
+                ecstore.clone(),
+                OffsetDateTime::now_utc(),
+                StdDuration::from_secs(24 * 60 * 60),
+            )
+            .with_subscriber(subscriber)
+            .await;
+
+            for (path, expected) in physical_paths.iter().zip(&original) {
+                assert_eq!(fs::read(path.join(STORAGE_FORMAT_FILE)).await.unwrap(), *expected);
+            }
+            let log_output = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap();
+            assert!(
+                log_output.contains(&upload_path),
+                "missing background diagnostic for {upload_path}: {log_output}"
+            );
+            assert!(log_output.contains("multipart_cleanup_lock_or_recheck_failed"), "{log_output}");
+            assert!(!log_output.contains("op=\"upload_part\""), "unexpected upload-part warning: {log_output}");
+            for path in physical_paths.iter().take(copies) {
+                fs::remove_dir_all(path).await.unwrap();
+            }
+        }
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    #[serial]
+    async fn stale_multipart_cleanup_lists_parts_only_when_size_is_needed() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        use tracing::instrument::WithSubscriber as _;
+        use tracing_subscriber::prelude::*;
+
+        struct PartListCounter(Arc<AtomicUsize>);
+        impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for PartListCounter {
+            fn on_new_span(
+                &self,
+                attrs: &tracing::span::Attributes<'_>,
+                _id: &tracing::span::Id,
+                _ctx: tracing_subscriber::layer::Context<'_, S>,
+            ) {
+                if attrs.metadata().name() == "list_object_parts" {
+                    self.0.fetch_add(1, Ordering::Relaxed);
+                }
+            }
+        }
+
+        let (_, ecstore) = setup_test_env().await;
+        let bucket = format!("stale-list-parts-{}", Uuid::new_v4().simple());
+        create_test_bucket(&ecstore, &bucket).await;
+        set_abort_incomplete_lifecycle(&bucket, "logs/", 0).await;
+        let now = OffsetDateTime::now_utc();
+        for (object, age, size_filter, should_list) in [
+            ("logs/default-expired", time::Duration::hours(30), false, false),
+            ("logs/lifecycle-only", time::Duration::minutes(5), false, false),
+            ("logs/size-filter", time::Duration::minutes(5), true, true),
+        ] {
+            if size_filter {
+                set_abort_incomplete_lifecycle_with_size(&bucket, "logs/", 0, 0).await;
+            }
+            let upload = ecstore
+                .new_multipart_upload(
+                    &bucket,
+                    object,
+                    &ObjectOptions {
+                        mod_time: Some(now - age),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("create multipart upload");
+            if size_filter {
+                let mut data = PutObjReader::from_vec(vec![1, 2, 3, 4]);
+                ecstore
+                    .put_object_part(&bucket, object, &upload.upload_id, 1, &mut data, &ObjectOptions::default())
+                    .await
+                    .expect("upload a part for size filtering");
+            }
+            let calls = Arc::new(AtomicUsize::new(0));
+            let subscriber = tracing_subscriber::registry().with(PartListCounter(Arc::clone(&calls)));
+            cleanup_stale_multipart_uploads_once_at(ecstore.clone(), now, StdDuration::from_secs(24 * 60 * 60))
+                .with_subscriber(subscriber)
+                .await;
+            let listed = calls.load(Ordering::Relaxed);
+            assert_eq!(listed > 0, should_list, "unexpected ListParts calls for {object}: {listed}");
+            let result = ecstore
+                .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
+                .await;
+            assert!(
+                matches!(result, Err(ref err) if is_err_invalid_upload_id(err)),
+                "upload must be removed: {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn stale_multipart_cleanup_rejects_incomplete_or_overflowing_part_sizes() {
+        use crate::storage_api_contracts::multipart::{ListPartsInfo, PartInfo};
+
+        let part = |size, actual_size, etag: &str| PartInfo {
+            size,
+            actual_size,
+            etag: Some(etag.to_owned()),
+            ..Default::default()
+        };
+        let mut parts = ListPartsInfo {
+            parts: vec![part(4, 6, "valid")],
+            ..Default::default()
+        };
+        assert_eq!(super::stale_upload_size_from_parts(&parts), Some(6));
+        parts.is_truncated = true;
+        assert_eq!(super::stale_upload_size_from_parts(&parts), None);
+        parts.is_truncated = false;
+        parts.parts = vec![part(0, 0, "")];
+        assert_eq!(super::stale_upload_size_from_parts(&parts), None);
+        parts.parts = vec![part(usize::MAX, 0, "valid"), part(1, 1, "valid")];
+        assert_eq!(super::stale_upload_size_from_parts(&parts), None);
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn stale_multipart_cleanup_handles_unknown_size_without_false_abort() {
+        let (_, ecstore) = setup_test_env().await;
+        let bucket = format!("stale-unknown-size-{}", Uuid::new_v4().simple());
+        let object = "logs/object";
+        create_test_bucket(&ecstore, &bucket).await;
+        let size_rule = r#"<Rule><ID>size</ID><Status>Enabled</Status><Filter><ObjectSizeLessThan>5</ObjectSizeLessThan></Filter><AbortIncompleteMultipartUpload><DaysAfterInitiation>0</DaysAfterInitiation></AbortIncompleteMultipartUpload></Rule>"#;
+        let independent_rule = r#"<Rule><ID>prefix</ID><Status>Enabled</Status><Filter><Prefix>logs/</Prefix></Filter><AbortIncompleteMultipartUpload><DaysAfterInitiation>0</DaysAfterInitiation></AbortIncompleteMultipartUpload></Rule>"#;
+        let lifecycle = |rules: &str| format!("<LifecycleConfiguration>{rules}</LifecycleConfiguration>").into_bytes();
+        metadata_sys::update(&bucket, BUCKET_LIFECYCLE_CONFIG, lifecycle(size_rule))
+            .await
+            .expect("install upper-size lifecycle rule");
+        let metadata = HashMap::from([
+            (super::RUSTFS_MULTIPART_BUCKET_KEY.to_owned(), bucket.clone()),
+            (super::RUSTFS_MULTIPART_OBJECT_KEY.to_owned(), object.to_owned()),
+        ]);
+        let initiated = OffsetDateTime::now_utc() - time::Duration::hours(1);
+        let set = &ecstore.pools[0].disk_set[0];
+        let upload = ecstore
+            .new_multipart_upload(&bucket, object, &ObjectOptions::default())
+            .await
+            .expect("create empty multipart upload");
+        let decoded = base64_simd::URL_SAFE_NO_PAD
+            .decode_to_vec(upload.upload_id.as_bytes())
+            .expect("decode upload ID");
+        let decoded = String::from_utf8(decoded).expect("upload ID must be UTF-8");
+        let (_, upload_dir) = decoded.split_once('.').expect("deployment-prefixed upload ID");
+        assert!(
+            super::stale_upload_lifecycle_due(set, &metadata, initiated, upload_dir, false)
+                .await
+                .is_some(),
+            "a known empty upload must match the upper-size rule"
+        );
+        let missing_upload = format!("{}x{}", Uuid::new_v4(), initiated.unix_timestamp_nanos());
+        assert!(
+            super::stale_upload_lifecycle_due(set, &metadata, initiated, &missing_upload, false)
+                .await
+                .is_none(),
+            "an unreadable size must not match the upper-size rule"
+        );
+        metadata_sys::update(&bucket, BUCKET_LIFECYCLE_CONFIG, lifecycle(&format!("{size_rule}{independent_rule}")))
+            .await
+            .expect("install size-independent lifecycle rule");
+        assert!(
+            super::stale_upload_lifecycle_due(set, &metadata, initiated, &missing_upload, false)
+                .await
+                .is_some(),
+            "a size-independent rule remains eligible when size is unknown"
+        );
+    }
+
     #[tokio::test]
     #[serial]
     async fn stale_multipart_cleanup_uses_default_expiry_without_lifecycle() {

crates/ecstore/src/config/com.rs

@@ -488,6 +488,18 @@ where
     Ok(data)
 }
 
+/// Read config bytes and metadata while preserving an existing empty payload.
+pub(crate) async fn read_config_preserve_empty_with_metadata_opts<S>(
+    api: Arc<S>,
+    file: &str,
+    opts: &ObjectOptions,
+) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: EcstoreObjectIO,
+{
+    read_config_with_metadata_inner(api, file, opts, true, None).await
+}
+
 pub async fn read_config_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
 where
     S: ObjectIO<

crates/ecstore/src/disk/local.rs

@@ -8450,6 +8450,23 @@ impl LocalDisk {
                     return Err(std::io::Error::other("metadata read worker cancelled"));
                 }
                 let _io_permit = io_permit;
+                #[cfg(all(any(test, feature = "test-util"), not(windows)))]
+                {
+                    let opened = root.open_readonly(&path);
+                    if opened.is_ok() {
+                        os::prepared_publication_test_hooks::run(
+                            os::prepared_publication_test_hooks::Stage::MetadataReadAfterOpen,
+                            &path,
+                        );
+                    }
+                    let result = read_fn(opened);
+                    os::prepared_publication_test_hooks::run(
+                        os::prepared_publication_test_hooks::Stage::MetadataReadAfterRead,
+                        &path,
+                    );
+                    Ok(result)
+                }
+                #[cfg(not(all(any(test, feature = "test-util"), not(windows))))]
                 Ok(read_fn(root.open_readonly(&path)))
             })
             .await

crates/ecstore/src/disk/os.rs

@@ -333,6 +333,8 @@ pub(crate) mod prepared_publication_test_hooks {
         PreparedRename,
         Rename,
         Remove,
+        MetadataReadAfterOpen,
+        MetadataReadAfterRead,
         #[cfg(test)]
         Rollback,
         #[cfg(test)]
@@ -457,6 +459,8 @@ pub enum LocalPublicationStage {
     PreparedRename,
     Rename,
     Remove,
+    MetadataReadAfterOpen,
+    MetadataReadAfterRead,
 }
 
 #[cfg(all(feature = "test-util", not(windows)))]
@@ -469,6 +473,8 @@ impl LocalPublicationPause {
             LocalPublicationStage::PreparedRename => prepared_publication_test_hooks::Stage::PreparedRename,
             LocalPublicationStage::Rename => prepared_publication_test_hooks::Stage::Rename,
             LocalPublicationStage::Remove => prepared_publication_test_hooks::Stage::Remove,
+            LocalPublicationStage::MetadataReadAfterOpen => prepared_publication_test_hooks::Stage::MetadataReadAfterOpen,
+            LocalPublicationStage::MetadataReadAfterRead => prepared_publication_test_hooks::Stage::MetadataReadAfterRead,
         };
         let (entered_tx, entered) = oneshot::channel();
         let (release, release_rx) = std::sync::mpsc::channel::<()>();

crates/ecstore/src/set_disk/mod.rs

@@ -4245,14 +4245,23 @@ pub(super) struct MultipartWriteQuorumContext<'a> {
     part_number: Option<usize>,
 }
 
+fn record_multipart_write_quorum_failure(
+    stage: &'static str,
+    errs: &[Option<DiskError>],
+    write_quorum: usize,
+) -> crate::disk::error_reduce::WriteQuorumFailureSummary {
+    let summary = build_write_quorum_failure_summary(errs, OBJECT_OP_IGNORED_ERRS, write_quorum);
+    runtime_sources::record_erasure_write_quorum_failure(stage, summary.dominant_error_label);
+    summary
+}
+
 fn log_multipart_write_quorum_failure(
     context: MultipartWriteQuorumContext<'_>,
     errs: &[Option<DiskError>],
     write_quorum: usize,
     returned_error: &DiskError,
 ) {
-    let summary = build_write_quorum_failure_summary(errs, OBJECT_OP_IGNORED_ERRS, write_quorum);
-    runtime_sources::record_erasure_write_quorum_failure(context.stage, summary.dominant_error_label);
+    let summary = record_multipart_write_quorum_failure(context.stage, errs, write_quorum);
     warn!(
         target: "rustfs_ecstore::set_disk",
         event = EVENT_SET_DISK_MULTIPART,

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -44,9 +44,9 @@ use super::super::{
     create_bitrot_writer, debug, disk, error, file_info_is_valid_for_metadata, get_complete_multipart_md5, get_header_map,
     get_str, insert_str, is_err_object_not_found, is_err_version_not_found, is_min_allowed_part_size,
     log_multipart_write_quorum_failure, parts_after_marker, path_join_buf, prepare_complete_multipart_parts,
-    record_compression_total_memory, reduce_read_quorum_errs, reduce_write_quorum_errs, remove_header_map, resolve_write_layout,
-    restore_commit_operation_id_from_metadata, should_persist_encryption_original_size, strip_internal_multipart_metadata,
-    to_object_err, warn,
+    record_compression_total_memory, record_multipart_write_quorum_failure, reduce_read_quorum_errs, reduce_write_quorum_errs,
+    remove_header_map, resolve_write_layout, restore_commit_operation_id_from_metadata, should_persist_encryption_original_size,
+    strip_internal_multipart_metadata, to_object_err, warn,
 };
 use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
 #[cfg(test)]
@@ -233,6 +233,19 @@ fn admitted_multipart_size(current: u64, candidate: u64, limit: u64) -> Result<u
     Ok(total)
 }
 
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum MultipartMetadataQuorum {
+    Read,
+    Write,
+    Cleanup,
+}
+
+impl From<bool> for MultipartMetadataQuorum {
+    fn from(write: bool) -> Self {
+        if write { Self::Write } else { Self::Read }
+    }
+}
+
 pub(crate) struct StaleMultipartCleanupGuard {
     file_info: FileInfo,
     upload_path: String,
@@ -1139,7 +1152,7 @@ impl SetDisks {
             }
             let upload_id = runtime_sources::deployment_upload_id(raw_upload_id);
             let file_info = match self
-                .check_multipart_upload_path_exists(bucket, object, &upload_id, &upload_path, false, true)
+                .check_multipart_upload_path_exists(bucket, object, &upload_id, &upload_path, MultipartMetadataQuorum::Read, true)
                 .await
             {
                 Ok((file_info, _)) => file_info,
@@ -1226,7 +1239,7 @@ impl SetDisks {
             .acquire_write_lock_diag("abort_data_movement_multipart", RUSTFS_META_MULTIPART_BUCKET, &upload_path)
             .await?;
         let file_info = match self
-            .check_multipart_upload_path_exists(bucket, object, upload_id, &upload_path, true, true)
+            .check_multipart_upload_path_exists(bucket, object, upload_id, &upload_path, MultipartMetadataQuorum::Write, true)
             .await
         {
             Ok((file_info, _)) => file_info,
@@ -1333,7 +1346,7 @@ impl SetDisks {
         data_movement: bool,
     ) -> Result<(FileInfo, Vec<FileInfo>)> {
         let upload_id_path = Self::validated_multipart_upload_path(bucket, object, upload_id, data_movement)?;
-        self.check_multipart_upload_path_exists(bucket, object, upload_id, &upload_id_path, write, false)
+        self.check_multipart_upload_path_exists(bucket, object, upload_id, &upload_id_path, write.into(), false)
             .await
     }
 
@@ -1373,7 +1386,7 @@ impl SetDisks {
         object: &str,
         upload_id: &str,
         upload_id_path: &str,
-        write: bool,
+        quorum_mode: MultipartMetadataQuorum,
         include_recovery: bool,
     ) -> Result<(FileInfo, Vec<FileInfo>)> {
         // Slow metadata I/O must not hold the topology registry's read lock.
@@ -1489,22 +1502,26 @@ impl SetDisks {
         }
 
         let mut quorum = read_quorum as usize;
-        if write {
+        if quorum_mode != MultipartMetadataQuorum::Read {
             quorum = write_quorum as usize;
 
             if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, quorum) {
-                log_multipart_write_quorum_failure(
-                    MultipartWriteQuorumContext {
-                        stage: MULTIPART_WRITE_QUORUM_UPLOAD_METADATA,
-                        bucket,
-                        object,
-                        upload_id,
-                        part_number: None,
-                    },
-                    &errs,
-                    quorum,
-                    &err,
-                );
+                if quorum_mode == MultipartMetadataQuorum::Write {
+                    log_multipart_write_quorum_failure(
+                        MultipartWriteQuorumContext {
+                            stage: MULTIPART_WRITE_QUORUM_UPLOAD_METADATA,
+                            bucket,
+                            object,
+                            upload_id,
+                            part_number: None,
+                        },
+                        &errs,
+                        quorum,
+                        &err,
+                    );
+                } else {
+                    record_multipart_write_quorum_failure(MULTIPART_WRITE_QUORUM_UPLOAD_METADATA, &errs, quorum);
+                }
                 return Err(map_error(err));
             }
         } else if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, quorum) {
@@ -1523,7 +1540,7 @@ impl SetDisks {
             .acquire_write_lock_diag("stale_multipart_cleanup", RUSTFS_META_MULTIPART_BUCKET, upload_path)
             .await?;
         let (file_info, _) = self
-            .check_multipart_upload_path_exists("", "", "", upload_path, true, true)
+            .check_multipart_upload_path_exists("", "", "", upload_path, MultipartMetadataQuorum::Cleanup, true)
             .await?;
         let write_quorum = file_info.write_quorum(self.default_write_quorum());
         Ok(StaleMultipartCleanupGuard {
@@ -3032,7 +3049,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         // its full write quorum and the upload's incarnation/lifecycle fences.
         let upload_id_path = Self::validated_multipart_upload_path(bucket, object, upload_id, opts.data_movement)?;
         let (fi, _) = self
-            .check_multipart_upload_path_exists(bucket, object, upload_id, &upload_id_path, false, true)
+            .check_multipart_upload_path_exists(bucket, object, upload_id, &upload_id_path, MultipartMetadataQuorum::Read, true)
             .await?;
         ensure_data_movement_upload_access(&fi, bucket, object, upload_id, opts)?;
         ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
@@ -3192,7 +3209,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     }
                     SourceRecovery::Missing => {
                         return match self
-                            .check_multipart_upload_path_exists(bucket, object, upload_id, &upload_id_path, false, true)
+                            .check_multipart_upload_path_exists(
+                                bucket,
+                                object,
+                                upload_id,
+                                &upload_id_path,
+                                MultipartMetadataQuorum::Read,
+                                true,
+                            )
                             .await
                         {
                             Ok(_) => Err(Error::ErasureWriteQuorum),
@@ -11133,7 +11157,7 @@ mod tests {
             *disk = None;
         }
         let (info, _) = set
-            .check_multipart_upload_path_exists(bucket, object, &upload, &path, false, true)
+            .check_multipart_upload_path_exists(bucket, object, &upload, &path, MultipartMetadataQuorum::Read, true)
             .await
             .expect("four primary plus four recovery copies reach metadata read quorum");
         assert_eq!(info.name, path);
@@ -11154,7 +11178,7 @@ mod tests {
             *disk = None;
         }
         assert!(
-            set.check_multipart_upload_path_exists(bucket, object, &upload, &path, false, true)
+            set.check_multipart_upload_path_exists(bucket, object, &upload, &path, MultipartMetadataQuorum::Read, true)
                 .await
                 .is_err(),
             "two files on each of four disks must not count as eight voters"
@@ -14471,7 +14495,7 @@ mod tests {
                     object,
                     &u_new,
                     &SetDisks::get_upload_id_dir(bucket, object, &u_new),
-                    true,
+                    MultipartMetadataQuorum::Write,
                     true,
                 )
                 .await
@@ -14516,7 +14540,7 @@ mod tests {
                 object,
                 &upload,
                 &SetDisks::get_upload_id_dir(bucket, object, &upload),
-                true,
+                MultipartMetadataQuorum::Write,
                 true,
             )
             .await
@@ -14599,7 +14623,7 @@ mod tests {
                     assert_completion_files_preserved(&disk_stores, RUSTFS_META_MULTIPART_BUCKET, &source_files).await;
                     set_disks
                         .check_multipart_upload_path_exists(
-                            bucket, object, &u_new, &SetDisks::get_upload_id_dir(bucket, object, &u_new), true, true,
+                            bucket, object, &u_new, &SetDisks::get_upload_id_dir(bucket, object, &u_new), MultipartMetadataQuorum::Write, true,
                         )
                         .await
                         .expect("upload authority must survive the post-commit crash");

crates/ecstore/src/set_disk/ops/multipart_abort_tests.rs

@@ -507,7 +507,7 @@ mod native_abort_closure_tests {
         let path = SetDisks::get_upload_id_dir(bucket, object, &upload);
         pin_historical_unmarked_upload(&disks, bucket, &path).await;
         let before = originals(&disks, &path).await;
-        set.check_multipart_upload_path_exists(bucket, object, &upload, &path, false, true)
+        set.check_multipart_upload_path_exists(bucket, object, &upload, &path, MultipartMetadataQuorum::Read, true)
             .await
             .expect("readonly legacy cleanup identity lookup");
         assert_eq!(
@@ -786,7 +786,7 @@ mod native_abort_closure_tests {
             "known orphan native evidence remains unknown: {inspected:?}"
         );
         let readonly = set
-            .check_multipart_upload_path_exists(bucket, object, &upload, &path, false, true)
+            .check_multipart_upload_path_exists(bucket, object, &upload, &path, MultipartMetadataQuorum::Read, true)
             .await;
         assert!(
             matches!(

crates/iam/src/delegation.rs

@@ -0,0 +1,313 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+
+//! Durable restrictions owned by an authenticated user. A record is not an IAM
+//! grant or credential; issuing credentials still requires current IAM checks.
+use crate::{error::is_err_config_not_found, store::Store};
+use serde::{Deserialize, Serialize};
+
+const PREFIX: &str = "config/iam/plugin-delegations/v1";
+
+#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct Restrictions {
+    #[serde(default, skip_serializing_if = "Option::is_none", deserialize_with = "present_value")]
+    pub version: Option<u32>,
+    #[serde(default, skip_serializing_if = "Option::is_none", deserialize_with = "present_value")]
+    pub output_bucket: Option<String>,
+    pub deployment: String,
+    pub bucket: String,
+    pub input_prefix: String,
+    pub output_prefix: String,
+    pub expires_at_unix_seconds: u64,
+}
+
+// An absent v1 field is compatible; an explicitly null v2 field is invalid.
+fn present_value<'de, T: Deserialize<'de>, D: serde::Deserializer<'de>>(deserializer: D) -> Result<Option<T>, D::Error> {
+    T::deserialize(deserializer).map(Some)
+}
+
+impl Restrictions {
+    pub fn output_bucket(&self) -> Result<&str, DelegationError> {
+        match (self.version, self.output_bucket.as_deref()) {
+            (None, None) => Ok(&self.bucket),
+            (Some(2), Some(output)) => Ok(output),
+            _ => Err(DelegationError::Invalid),
+        }
+    }
+
+    pub fn validate(&self, now: u64) -> Result<(), DelegationError> {
+        let output_bucket = self.output_bucket()?;
+        let names: [&str; 5] = [
+            &self.deployment,
+            &self.bucket,
+            &self.input_prefix,
+            &self.output_prefix,
+            output_bucket,
+        ];
+        if names
+            .iter()
+            .any(|s| s.is_empty() || s.len() > 1024 || s.contains("${") || s.chars().any(char::is_control))
+            || self.bucket.contains('/')
+            || self.bucket.contains('*')
+            || self.bucket.contains('?')
+            || output_bucket.contains('/')
+            || output_bucket.contains('*')
+            || output_bucket.contains('?')
+            || self.input_prefix.contains('*')
+            || self.input_prefix.contains('?')
+            || self.output_prefix.contains('*')
+            || self.output_prefix.contains('?')
+            || self.expires_at_unix_seconds <= now
+            || self.expires_at_unix_seconds.saturating_sub(now) > 86400
+        {
+            return Err(DelegationError::Invalid);
+        }
+        Ok(())
+    }
+}
+
+#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct Delegation {
+    pub version: u32,
+    pub id: String,
+    pub revision: String,
+    pub owner_access_key: String,
+    pub created_at_unix_seconds: u64,
+    pub restrictions: Restrictions,
+}
+
+#[derive(Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+struct Tombstone {
+    version: u32,
+    id: String,
+    revision: String,
+    owner_access_key: String,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum DelegationError {
+    #[error("invalid delegation restriction")]
+    Invalid,
+    #[error("delegation not found")]
+    NotFound,
+    #[error("delegation owner rejected")]
+    OwnerRejected,
+    #[error("delegation unavailable")]
+    Unavailable,
+    #[error("delegation revoked")]
+    Revoked,
+    #[error("delegation expired")]
+    Expired,
+}
+
+fn record_path(id: &str) -> Result<String, DelegationError> {
+    if id.len() != 32 || !id.bytes().all(|c| c.is_ascii_hexdigit()) {
+        return Err(DelegationError::Invalid);
+    }
+    Ok(format!("{PREFIX}/{id}/record.json"))
+}
+
+async fn load<T: Store>(store: &T, id: &str) -> Result<Delegation, DelegationError> {
+    let record: Delegation = store.load_iam_config_strict(record_path(id)?).await.map_err(|error| {
+        if is_err_config_not_found(&error) {
+            DelegationError::NotFound
+        } else {
+            DelegationError::Unavailable
+        }
+    })?;
+    if !matches!(record.version, 1 | 2)
+        || record.version != record.restrictions.version.unwrap_or(1)
+        || record.id != id
+        || record.revision != id
+        || record.owner_access_key.is_empty()
+    {
+        return Err(DelegationError::Unavailable);
+    }
+    record
+        .restrictions
+        .validate(record.created_at_unix_seconds)
+        .map_err(|_| DelegationError::Unavailable)?;
+    Ok(record)
+}
+
+pub(crate) async fn create<T: Store>(
+    store: &T,
+    owner: &str,
+    restrictions: Restrictions,
+    now: u64,
+) -> Result<Delegation, DelegationError> {
+    restrictions.validate(now)?;
+    if owner.is_empty() {
+        return Err(DelegationError::OwnerRejected);
+    }
+    // IDs are generated exclusively here; no update operation accepts an ID.
+    let id = format!("{:032x}", rand::random::<u128>());
+    let path = record_path(&id)?;
+    match store.load_iam_config_strict::<Delegation>(&path).await {
+        Err(error) if is_err_config_not_found(&error) => {}
+        _ => return Err(DelegationError::Unavailable),
+    }
+    let record = Delegation {
+        version: restrictions.version.unwrap_or(1),
+        revision: id.clone(),
+        id,
+        owner_access_key: owner.to_owned(),
+        created_at_unix_seconds: now,
+        restrictions,
+    };
+    store
+        .save_iam_config(&record, path)
+        .await
+        .map_err(|_| DelegationError::Unavailable)?;
+    if load(store, &record.id).await? != record {
+        return Err(DelegationError::Unavailable);
+    }
+    Ok(record)
+}
+
+pub(crate) async fn revoke<T: Store>(store: &T, owner: &str, id: &str) -> Result<(), DelegationError> {
+    let record = load(store, id).await?;
+    if record.owner_access_key != owner {
+        return Err(DelegationError::OwnerRejected);
+    }
+    let tombstone = Tombstone {
+        version: 1,
+        id: record.id,
+        revision: record.revision,
+        owner_access_key: record.owner_access_key,
+    };
+    let path = format!("{PREFIX}/{id}/revoked.json");
+    // Every retry writes exactly the same revocation; active records never change.
+    store
+        .save_iam_config(&tombstone, &path)
+        .await
+        .map_err(|_| DelegationError::Unavailable)?;
+    let saved: Tombstone = store
+        .load_iam_config_strict(path)
+        .await
+        .map_err(|_| DelegationError::Unavailable)?;
+    if saved != tombstone {
+        return Err(DelegationError::Unavailable);
+    }
+    Ok(())
+}
+
+pub(crate) async fn read_active<T: Store>(store: &T, owner: &str, id: &str, now: u64) -> Result<Delegation, DelegationError> {
+    let record = load(store, id).await?;
+    if record.owner_access_key != owner {
+        return Err(DelegationError::OwnerRejected);
+    }
+    check_active(store, record, now).await
+}
+
+pub(crate) async fn read_owned<T: Store>(store: &T, owner: &str, id: &str) -> Result<Delegation, DelegationError> {
+    let record = load(store, id).await?;
+    if record.owner_access_key != owner {
+        return Err(DelegationError::OwnerRejected);
+    }
+    Ok(record)
+}
+
+pub(crate) async fn read_active_for_broker<T: Store>(store: &T, id: &str, now: u64) -> Result<Delegation, DelegationError> {
+    let record = load(store, id).await?;
+    check_active(store, record, now).await
+}
+async fn check_active<T: Store>(store: &T, record: Delegation, now: u64) -> Result<Delegation, DelegationError> {
+    let id = &record.id;
+    match store
+        .load_iam_config_strict::<Tombstone>(format!("{PREFIX}/{id}/revoked.json"))
+        .await
+    {
+        Ok(_) => return Err(DelegationError::Revoked),
+        Err(error) if is_err_config_not_found(&error) => {}
+        Err(_) => return Err(DelegationError::Unavailable),
+    }
+    if now >= record.restrictions.expires_at_unix_seconds {
+        return Err(DelegationError::Expired);
+    }
+    Ok(record)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn delegation_v1_json_remains_compatible() {
+        let legacy = serde_json::json!({"deployment":"runner-a","bucket":"photos",
+            "input_prefix":"inputs/","output_prefix":"outputs/","expires_at_unix_seconds":200});
+        let restriction: Restrictions = serde_json::from_value(legacy.clone()).expect("legacy restriction");
+        restriction.validate(100).expect("legacy validation");
+        assert_eq!(restriction.output_bucket().expect("v1 output"), "photos");
+        assert_eq!(serde_json::to_value(restriction).expect("legacy serialization"), legacy);
+        let record = serde_json::json!({"version":1,"id":"a".repeat(32),"revision":"a".repeat(32),
+            "owner_access_key":"owner","created_at_unix_seconds":100,"restrictions":legacy});
+        let decoded: Delegation = serde_json::from_value(record.clone()).expect("legacy record");
+        assert_eq!(serde_json::to_value(decoded).expect("legacy record serialization"), record);
+    }
+
+    #[test]
+    fn delegation_v2_requires_explicit_output_bucket() {
+        let valid = serde_json::json!({"version":2,"deployment":"runner-a","bucket":"photos",
+            "input_prefix":"inputs/","output_bucket":"artifacts","output_prefix":"outputs/",
+            "expires_at_unix_seconds":200});
+        let restriction: Restrictions = serde_json::from_value(valid.clone()).expect("v2 restriction");
+        restriction.validate(100).expect("explicit output bucket");
+        assert_eq!(restriction.output_bucket().expect("v2 output"), "artifacts");
+        assert_eq!(serde_json::to_value(restriction).expect("v2 serialization"), valid);
+        for key in ["version", "output_bucket"] {
+            let mut bad = valid.clone();
+            bad.as_object_mut().expect("object").remove(key);
+            assert!(
+                serde_json::from_value::<Restrictions>(bad).map_or(true, |r| r.validate(100).is_err()),
+                "v2 required {key}"
+            );
+        }
+        for (key, value) in [
+            ("version", serde_json::json!(1)),
+            ("version", serde_json::json!(3)),
+            ("version", serde_json::Value::Null),
+            ("output_bucket", serde_json::Value::Null),
+            ("output_bucket", serde_json::json!("")),
+            ("output_bucket", serde_json::json!("other/*")),
+            ("output_bucket", serde_json::json!("${aws:username}")),
+            ("output_bucket", serde_json::json!("other?")),
+            ("output_bucket", serde_json::json!("other\n")),
+            ("unknown", serde_json::json!(true)),
+        ] {
+            let mut bad = valid.clone();
+            bad[key] = value;
+            assert!(
+                serde_json::from_value::<Restrictions>(bad).map_or(true, |r| r.validate(100).is_err()),
+                "invalid v2 {key}"
+            );
+        }
+    }
+
+    #[test]
+    fn restrictions_are_literal_bounded_and_expiring() {
+        let mut restriction = Restrictions {
+            version: None,
+            output_bucket: None,
+            deployment: "runner-a".into(),
+            bucket: "photos".into(),
+            input_prefix: "inputs/".into(),
+            output_prefix: "outputs/".into(),
+            expires_at_unix_seconds: 200,
+        };
+        assert!(restriction.validate(100).is_ok());
+        restriction.output_prefix = "outputs/*".into();
+        assert!(matches!(restriction.validate(100), Err(DelegationError::Invalid)));
+        restriction.output_prefix = "outputs/".into();
+        assert!(matches!(restriction.validate(200), Err(DelegationError::Invalid)));
+        restriction.expires_at_unix_seconds = 200;
+        restriction.input_prefix = "inputs/${aws:username}/".into();
+        assert!(
+            matches!(restriction.validate(100), Err(DelegationError::Invalid)),
+            "C07_LITERAL_SCOPE_BOUND: policy variables are not literal restrictions"
+        );
+        assert!(record_path("../../users").is_err());
+    }
+}

crates/iam/src/lib.rs

@@ -28,6 +28,7 @@ pub(crate) const EVENT_OIDC_DIAGNOSTICS: &str = "oidc_diagnostics";
 const EVENT_OIDC_STATE: &str = "oidc_state";
 
 pub mod cache;
+pub mod delegation;
 pub mod error;
 pub mod federation;
 pub mod keyring;

crates/iam/src/storage_api.rs

@@ -91,6 +91,7 @@ pub(crate) mod crate_boundary {
 
 pub(crate) mod object_store {
     pub(crate) use super::storage_contracts::{HTTPPreconditions, ListOperations, ObjectInfoOrErr, ObjectOperations};
+    pub(crate) use rustfs_ecstore::api::config::com::read_config_preserve_empty_with_metadata_opts;
 }
 
 pub(crate) mod runtime {

crates/iam/src/store.rs

@@ -26,6 +26,11 @@ pub trait Store: Clone + Send + Sync + 'static {
     fn has_watcher(&self) -> bool;
     async fn save_iam_config<Item: Serialize + Send>(&self, item: Item, path: impl AsRef<str> + Send) -> Result<()>;
     async fn load_iam_config<Item: DeserializeOwned>(&self, path: impl AsRef<str> + Send) -> Result<Item>;
+    /// Only confirmed storage absence may return ConfigNotFound. Corrupt or
+    /// undecryptable objects must remain errors, especially for revocations.
+    async fn load_iam_config_strict<Item: DeserializeOwned>(&self, _path: impl AsRef<str> + Send) -> Result<Item> {
+        Err(crate::error::Error::IamSysNotInitialized)
+    }
     async fn delete_iam_config(&self, path: impl AsRef<str> + Send) -> Result<()>;
 
     async fn save_user_identity(&self, name: &str, user_type: UserType, item: UserIdentity, ttl: Option<usize>) -> Result<()>;

crates/iam/src/store/object.rs

@@ -42,6 +42,7 @@ use tracing::{debug, error, warn};
 
 use crate::storage_api::object_store::{
     HTTPPreconditions, ListOperations, ObjectInfoOrErr as StorageObjectInfoOrErr, ObjectOperations,
+    read_config_preserve_empty_with_metadata_opts,
 };
 
 pub static IAM_CONFIG_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam"));
@@ -611,6 +612,11 @@ impl ObjectStore {
         Ok(serde_json::from_slice(&outcome.plain)?)
     }
 
+    fn decode_iam_config_strict<Item: DeserializeOwned>(data: &[u8]) -> Result<Item> {
+        let outcome = Self::decrypt_data_with_source(data)?;
+        Ok(serde_json::from_slice(&outcome.plain)?)
+    }
+
     /// Parameterized core of [`Store::load_user_identity`].
     async fn load_user_identity_with(&self, name: &str, user_type: UserType, mode: LoadMode) -> Result<UserIdentity> {
         let mut u: UserIdentity = self
@@ -824,6 +830,12 @@ impl Store for ObjectStore {
     async fn load_iam_config<Item: DeserializeOwned>(&self, path: impl AsRef<str> + Send) -> Result<Item> {
         self.load_iam_config_with(path, LoadMode::Locked).await
     }
+    async fn load_iam_config_strict<Item: DeserializeOwned>(&self, path: impl AsRef<str> + Send) -> Result<Item> {
+        let (data, _) =
+            read_config_preserve_empty_with_metadata_opts(self.object_api.clone(), path.as_ref(), &LoadMode::Locked.read_opts())
+                .await?;
+        Self::decode_iam_config_strict(&data)
+    }
     /// Saves IAM configuration with a retry mechanism on failure.
     ///
     /// Attempts to save the IAM configuration up to 5 times if the storage layer is not ready,
@@ -1458,6 +1470,40 @@ mod tests {
         assert!(result.is_err(), "corrupt stream_io data should fail decrypt");
     }
 
+    #[test]
+    fn strict_revocation_load_keeps_real_crypto_failure_distinct_from_absence() {
+        let cred = test_cred();
+        let key = format!("{}:{}", cred.access_key, cred.secret_key);
+        let plain = br#"{"version":1,"id":"revoked"}"#;
+        let mut encrypted = rustfs_crypto::encrypt_stream_io(key.as_bytes(), plain).expect("real stored ciphertext");
+        // Exercise the exact decoder used after the actual object read.
+        assert!(ObjectStore::decode_iam_config_strict::<serde_json::Value>(&encrypted).is_ok());
+        encrypted[50] ^= 0xFF;
+        let error = ObjectStore::decode_iam_config_strict::<serde_json::Value>(&encrypted)
+            .expect_err("corrupt tombstone must not become missing");
+        assert!(
+            matches!(error, Error::CryptoError(_)),
+            "C07_REVOCATION_FAIL_CLOSED: preserve actual crypto error"
+        );
+        assert!(
+            !crate::error::is_err_config_not_found(&error),
+            "C07_REVOCATION_FAIL_CLOSED: corrupted revocation is not absence"
+        );
+        // The preserve-empty reader keeps an existing zero-byte object as
+        // Ok(empty); ecstore's read_config_preserve_empty_distinguishes_empty_object
+        // covers that storage boundary. Its payload must fail here, not become
+        // ConfigNotFound and reactivate a revoked delegation.
+        let empty =
+            ObjectStore::decode_iam_config_strict::<serde_json::Value>(&[]).expect_err("an existing empty tombstone is corrupt");
+        assert!(
+            !crate::error::is_err_config_not_found(&empty),
+            "C07_REVOCATION_FAIL_CLOSED: empty existing revocation is not absence"
+        );
+        let invalid_json =
+            ObjectStore::decode_iam_config_strict::<serde_json::Value>(br#"{"version": }"#).expect_err("invalid stored JSON");
+        assert!(!crate::error::is_err_config_not_found(&invalid_json));
+    }
+
     #[test]
     fn test_decrypt_data_short_data_fails() {
         let short = &[0x00u8; 40]; // less than 41-byte stream_io header, not valid JSON

crates/iam/src/sys.rs

@@ -75,6 +75,7 @@ pub const STATUS_ENABLED: &str = "enabled";
 pub const STATUS_DISABLED: &str = "disabled";
 
 pub const POLICYNAME: &str = "policy";
+pub const PLUGIN_ATTEMPT_CREDENTIAL_CLAIM: &str = "rustfs:plugin-attempt-v1";
 pub const SESSION_POLICY_NAME: &str = "sessionPolicy";
 pub const SESSION_POLICY_NAME_EXTRACTED: &str = "sessionPolicy-extracted";
 const VERIFIED_FEDERATED_POLICY_CLAIM: &str = "x-rustfs-internal-federated-policy";
@@ -168,6 +169,12 @@ enum PreparedServicePolicyMode {
 
 #[derive(Clone)]
 enum PreparedIamMode {
+    TerminalDeny,
+    PluginAttemptSts {
+        is_owner: bool,
+        combined_policy: Policy,
+        session_policy: PreparedSessionPolicy,
+    },
     Opa,
     Owner,
     Deny,
@@ -202,7 +209,7 @@ impl PreparedIamAuth {
     pub async fn needs_existing_object_tag_for_args(&self, args: &Args<'_>) -> bool {
         match &self.mode {
             PreparedIamMode::Opa => true,
-            PreparedIamMode::Owner | PreparedIamMode::Deny => false,
+            PreparedIamMode::Owner | PreparedIamMode::Deny | PreparedIamMode::TerminalDeny => false,
             PreparedIamMode::Regular { combined_policy } => {
                 policy_needs_existing_object_tag_for_args(combined_policy, args).await
             }
@@ -210,6 +217,11 @@ impl PreparedIamAuth {
                 combined_policy,
                 session_policy,
                 ..
+            }
+            | PreparedIamMode::PluginAttemptSts {
+                combined_policy,
+                session_policy,
+                ..
             } => {
                 policy_needs_existing_object_tag_for_args(combined_policy, args).await
                     || prepared_session_policy_needs_existing_object_tag_for_args(session_policy, args).await
@@ -239,14 +251,56 @@ impl PreparedIamAuth {
     pub fn combined_policy_for_view(&self) -> Option<&Policy> {
         match &self.mode {
             PreparedIamMode::Regular { combined_policy } => Some(combined_policy),
-            PreparedIamMode::Sts { combined_policy, .. } => Some(combined_policy),
+            PreparedIamMode::Sts { combined_policy, .. } | PreparedIamMode::PluginAttemptSts { combined_policy, .. } => {
+                Some(combined_policy)
+            }
             PreparedIamMode::ServiceAccount { combined_policy, .. } => Some(combined_policy),
-            PreparedIamMode::Opa | PreparedIamMode::Owner | PreparedIamMode::Deny => None,
+            PreparedIamMode::Opa | PreparedIamMode::Owner | PreparedIamMode::Deny | PreparedIamMode::TerminalDeny => None,
         }
     }
 }
 
 impl<T: Store> IamSys<T> {
+    /// For the authenticated, explicitly configured broker handler only.
+    pub async fn read_active_plugin_delegation_for_broker(
+        &self,
+        id: &str,
+        now: u64,
+    ) -> std::result::Result<crate::delegation::Delegation, crate::delegation::DelegationError> {
+        crate::delegation::read_active_for_broker(&self.store.api, id, now).await
+    }
+    pub async fn create_plugin_delegation(
+        &self,
+        owner: &str,
+        restrictions: crate::delegation::Restrictions,
+        now: u64,
+    ) -> std::result::Result<crate::delegation::Delegation, crate::delegation::DelegationError> {
+        crate::delegation::create(&self.store.api, owner, restrictions, now).await
+    }
+    pub async fn revoke_plugin_delegation(
+        &self,
+        owner: &str,
+        id: &str,
+    ) -> std::result::Result<(), crate::delegation::DelegationError> {
+        crate::delegation::revoke(&self.store.api, owner, id).await
+    }
+    pub async fn read_active_plugin_delegation(
+        &self,
+        owner: &str,
+        id: &str,
+        now: u64,
+    ) -> std::result::Result<crate::delegation::Delegation, crate::delegation::DelegationError> {
+        crate::delegation::read_active(&self.store.api, owner, id, now).await
+    }
+    /// Reads an owned delegation record for historical lookups, including expired or revoked records.
+    pub async fn read_owned_plugin_delegation(
+        &self,
+        owner: &str,
+        id: &str,
+    ) -> std::result::Result<crate::delegation::Delegation, crate::delegation::DelegationError> {
+        crate::delegation::read_owned(&self.store.api, owner, id).await
+    }
+
     /// Create a new IamSys instance with the given IamCache store
     ///
     /// # Arguments
@@ -281,6 +335,16 @@ impl<T: Store> IamSys<T> {
         *guard = PolicyPluginState::Ready(client);
     }
 
+    /// Restricted credentials require the built-in parent/session intersection.
+    /// External authorization bypasses that path, so only Disabled is supported.
+    pub async fn ensure_restricted_credential_vending() -> Result<()> {
+        if matches!(Self::policy_plugin_state().await, PolicyPluginState::Disabled) {
+            Ok(())
+        } else {
+            Err(Error::IAMActionNotAllowed)
+        }
+    }
+
     pub async fn get_policy_plugin_client() -> Option<rustfs_policy::policy::opa::AuthZPlugin> {
         let policy_plugin_state = get_policy_plugin_state();
         let guard = policy_plugin_state.read().await;
@@ -1383,17 +1447,50 @@ impl<T: Store> IamSys<T> {
     /// 1) know whether policy evaluation may need `s3:ExistingObjectTag`, and
     /// 2) evaluate with final conditions without re-merging identity policies.
     pub async fn prepare_auth(&self, args: &Args<'_>) -> PreparedIamAuth {
-        if args.is_owner {
+        let marked = args.claims.contains_key(PLUGIN_ATTEMPT_CREDENTIAL_CLAIM);
+        if args.is_owner && !marked {
             return PreparedIamAuth {
                 needs_existing_object_tag: false,
                 mode: PreparedIamMode::Owner,
             };
         }
-
-        if let Some(prepared) = Self::policy_plugin_state().await.prepared_iam_auth() {
+        // One observed state decides this request's path. Marked credentials
+        // never enter OPA, including after a restart or on another node.
+        let state = Self::policy_plugin_state().await;
+        if marked {
+            if args.is_owner
+                || args.claims.get(PLUGIN_ATTEMPT_CREDENTIAL_CLAIM) != Some(&Value::Bool(true))
+                || !matches!(state, PolicyPluginState::Disabled)
+            {
+                return PreparedIamAuth {
+                    needs_existing_object_tag: false,
+                    mode: PreparedIamMode::TerminalDeny,
+                };
+            }
+            let prepared = self.prepare_builtin_auth(args).await;
+            let mode = match prepared.mode {
+                PreparedIamMode::Sts {
+                    is_owner,
+                    combined_policy,
+                    session_policy,
+                } => PreparedIamMode::PluginAttemptSts {
+                    is_owner,
+                    combined_policy,
+                    session_policy,
+                },
+                _ => PreparedIamMode::TerminalDeny,
+            };
+            return PreparedIamAuth {
+                needs_existing_object_tag: prepared.needs_existing_object_tag,
+                mode,
+            };
+        }
+        if let Some(prepared) = state.prepared_iam_auth() {
             return prepared;
         }
-
+        self.prepare_builtin_auth(args).await
+    }
+    async fn prepare_builtin_auth(&self, args: &Args<'_>) -> PreparedIamAuth {
         let Ok((is_svc, parent_user)) = self.is_service_account(args.account).await else {
             return PreparedIamAuth {
                 needs_existing_object_tag: false,
@@ -1425,6 +1522,20 @@ impl<T: Store> IamSys<T> {
     /// bucket policies on an IAM denial. An error must not become an implicit deny.
     pub async fn try_eval_prepared(&self, prepared: &PreparedIamAuth, args: &Args<'_>) -> Result<bool> {
         Ok(match &prepared.mode {
+            PreparedIamMode::TerminalDeny => return Err(Error::IAMActionNotAllowed),
+            PreparedIamMode::PluginAttemptSts {
+                is_owner,
+                combined_policy,
+                session_policy,
+            } => {
+                // A false IAM result permits bucket-policy fallback. Restricted
+                // plugin credentials must instead terminate authorization.
+                let session_allowed = evaluate_prepared_session_policy(session_policy, args).await == Some(true);
+                if !session_allowed || !(*is_owner || combined_policy.is_allowed(args).await) {
+                    return Err(Error::IAMActionNotAllowed);
+                }
+                true
+            }
             PreparedIamMode::Opa => {
                 let Some(opa_enable) = Self::get_policy_plugin_client().await else {
                     tracing::warn!("eval_prepared: OPA mode requested but plugin is unavailable");
@@ -2127,6 +2238,145 @@ mod tests {
         assert!(prepared.combined_policy_for_view().is_none());
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn marked_signed_sts_denial_is_terminal_before_public_bucket_fallback() {
+        let iam = IamSys::new(IamCache::new(StsTestMockStore::new(false)).await.unwrap());
+        let previous = IamSys::<StsTestMockStore>::policy_plugin_state().await;
+        let signing_key = "c07-test-token-signing-key";
+        let mut metadata=HashMap::from([
+            ("parent".into(),Value::String("sts-fallback-test-parent".into())),
+            ("exp".into(),serde_json::json!((OffsetDateTime::now_utc()+time::Duration::hours(1)).unix_timestamp())),
+            (SESSION_POLICY_NAME.into(),Value::String(base64_simd::URL_SAFE_NO_PAD.encode_to_string(br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::mybucket/inside"]}]}"#))),
+        ]);
+        metadata.insert(PLUGIN_ATTEMPT_CREDENTIAL_CLAIM.into(), Value::Bool(true));
+        let mut marked = get_new_credentials_with_metadata(&metadata, signing_key).unwrap();
+        marked.parent_user = "sts-fallback-test-parent".into();
+        let marked_claims: HashMap<String, Value> = rustfs_policy::utils::extract_claims(&marked.session_token, signing_key)
+            .expect("signed test credentials must decode")
+            .claims;
+        let marked_user = UserIdentity::from(marked.clone());
+        iam.store
+            .cache
+            .add_or_update_sts_account(&marked.access_key, &marked_user, OffsetDateTime::now_utc());
+        metadata.remove(PLUGIN_ATTEMPT_CREDENTIAL_CLAIM);
+        let mut ordinary = get_new_credentials_with_metadata(&metadata, signing_key).unwrap();
+        ordinary.parent_user = marked.parent_user.clone();
+        let ordinary_claims: HashMap<String, Value> = rustfs_policy::utils::extract_claims(&ordinary.session_token, signing_key)
+            .expect("signed test credentials must decode")
+            .claims;
+        iam.store.cache.add_or_update_sts_account(
+            &ordinary.access_key,
+            &UserIdentity::from(ordinary.clone()),
+            OffsetDateTime::now_utc(),
+        );
+        let groups = None;
+        let conditions = HashMap::new();
+        let args = Args {
+            account: &marked.access_key,
+            groups: &groups,
+            action: Action::S3Action(S3Action::GetObjectAction),
+            bucket: "mybucket",
+            object: "outside",
+            conditions: &conditions,
+            is_owner: false,
+            claims: &marked_claims,
+            deny_only: false,
+        };
+        *get_policy_plugin_state().write().await = PolicyPluginState::Disabled;
+        let denied = iam.prepare_auth(&args).await;
+        let disabled_terminal = matches!(iam.try_eval_prepared(&denied, &args).await, Err(Error::IAMActionNotAllowed));
+        let inside = Args {
+            object: "inside",
+            ..args.clone()
+        };
+        let allowed = iam.prepare_auth(&inside).await;
+        let inside_allowed = matches!(iam.try_eval_prepared(&allowed, &inside).await, Ok(true));
+        let unmarked_args = Args {
+            account: &ordinary.access_key,
+            claims: &ordinary_claims,
+            ..args.clone()
+        };
+        let unmarked = iam.prepare_auth(&unmarked_args).await;
+        let ordinary_false = matches!(iam.try_eval_prepared(&unmarked, &unmarked_args).await, Ok(false));
+        let public:rustfs_policy::policy::BucketPolicy=serde_json::from_value(serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":["*"]},"Action":["s3:GetObject"],"Resource":["arn:aws:s3:::mybucket/*"]}]})).unwrap();
+        let public_allows = public
+            .is_allowed(&rustfs_policy::policy::BucketPolicyArgs {
+                account: &marked.access_key,
+                groups: &groups,
+                action: args.action,
+                bucket: args.bucket,
+                object: args.object,
+                conditions: &conditions,
+                is_owner: false,
+            })
+            .await;
+        IamSys::<StsTestMockStore>::set_policy_plugin_client(opa::AuthZPlugin::new(opa::Args {
+            url: "http://127.0.0.1:8181/allow".into(),
+            auth_token: String::new(),
+        }))
+        .await;
+        let opa_marked = iam.prepare_auth(&args).await;
+        let ready_terminal = matches!(iam.try_eval_prepared(&opa_marked, &args).await, Err(Error::IAMActionNotAllowed));
+        let opa_ordinary = iam.prepare_auth(&unmarked_args).await;
+        let ordinary_keeps_opa = matches!(opa_ordinary.mode, PreparedIamMode::Opa);
+        // The captured Disabled preparation stays on built-in intersection even
+        // if an external policy client becomes ready before evaluation.
+        let captured_still_terminal = matches!(iam.try_eval_prepared(&denied, &args).await, Err(Error::IAMActionNotAllowed));
+        *get_policy_plugin_state().write().await = PolicyPluginState::Initializing;
+        let pending = iam.prepare_auth(&args).await;
+        let initializing_terminal = matches!(iam.try_eval_prepared(&pending, &args).await, Err(Error::IAMActionNotAllowed));
+        *get_policy_plugin_state().write().await = PolicyPluginState::Failed;
+        let failed = iam.prepare_auth(&args).await;
+        let failed_terminal = matches!(iam.try_eval_prepared(&failed, &args).await, Err(Error::IAMActionNotAllowed));
+        *get_policy_plugin_state().write().await = previous;
+        assert!(public_allows, "C07_PUBLIC_BUCKET_CONTROL: real public policy permits outside object");
+        assert!(
+            disabled_terminal && captured_still_terminal,
+            "C07_SCOPE_TERMINAL_BOUND: bucket policy must not rescue restricted scope denial"
+        );
+        assert!(
+            ready_terminal && initializing_terminal && failed_terminal,
+            "C07_EXTERNAL_POLICY_BOUND: marked persisted STS cannot enter OPA"
+        );
+        assert!(
+            inside_allowed && ordinary_false && ordinary_keeps_opa,
+            "C07_UNMARKED_COMPAT: preserve ordinary STS and valid marked scope"
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn restricted_vending_rejects_actual_external_policy_plugin_state() {
+        let previous = IamSys::<StsTestMockStore>::policy_plugin_state().await;
+        IamSys::<StsTestMockStore>::set_policy_plugin_client(opa::AuthZPlugin::new(opa::Args {
+            url: "http://127.0.0.1:8181/v1/data/rustfs/authz/allow".into(),
+            auth_token: String::new(),
+        }))
+        .await;
+        let active_rejected = IamSys::<StsTestMockStore>::ensure_restricted_credential_vending()
+            .await
+            .is_err();
+        *get_policy_plugin_state().write().await = PolicyPluginState::Initializing;
+        let initializing_rejected = IamSys::<StsTestMockStore>::ensure_restricted_credential_vending()
+            .await
+            .is_err();
+        *get_policy_plugin_state().write().await = PolicyPluginState::Failed;
+        let failed_rejected = IamSys::<StsTestMockStore>::ensure_restricted_credential_vending()
+            .await
+            .is_err();
+        *get_policy_plugin_state().write().await = PolicyPluginState::Disabled;
+        let disabled_allowed = IamSys::<StsTestMockStore>::ensure_restricted_credential_vending()
+            .await
+            .is_ok();
+        *get_policy_plugin_state().write().await = previous;
+        assert!(
+            active_rejected && initializing_rejected && failed_rejected,
+            "C07_EXTERNAL_POLICY_BOUND: external authorization cannot bypass minted session scope"
+        );
+        assert!(disabled_allowed);
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_prepare_auth_requests_existing_object_tags_in_opa_mode() {

crates/policy/src/policy/action.rs

@@ -568,6 +568,24 @@ pub enum AdminAction {
     GetTableLifecycleAction,
     #[strum(serialize = "admin:SetTableLifecycle")]
     SetTableLifecycleAction,
+    #[strum(serialize = "admin:CreatePluginDelegation")]
+    CreatePluginDelegationAction,
+    #[strum(serialize = "admin:IssuePluginAttempt")]
+    IssuePluginAttemptAction,
+    #[strum(serialize = "admin:SubmitPluginJob")]
+    SubmitPluginJobAction,
+    #[strum(serialize = "admin:GetPluginRun")]
+    GetPluginRunAction,
+    #[strum(serialize = "admin:CancelPluginRun")]
+    CancelPluginRunAction,
+    #[strum(serialize = "admin:ListPluginRuns")]
+    ListPluginRunsAction,
+    #[strum(serialize = "admin:ReplayPluginRun")]
+    ReplayPluginRunAction,
+    #[strum(serialize = "admin:GetPluginDelegation")]
+    GetPluginDelegationAction,
+    #[strum(serialize = "admin:RevokePluginDelegation")]
+    RevokePluginDelegationAction,
     #[strum(serialize = "admin:GetTableCredentials")]
     GetTableCredentialsAction,
     #[strum(serialize = "admin:RunTableMaintenance")]
@@ -713,6 +731,15 @@ impl AdminAction {
                 | AdminAction::GetTableLifecycleAction
                 | AdminAction::SetTableLifecycleAction
                 | AdminAction::GetTableCredentialsAction
+                | AdminAction::IssuePluginAttemptAction
+                | AdminAction::SubmitPluginJobAction
+                | AdminAction::GetPluginRunAction
+                | AdminAction::CancelPluginRunAction
+                | AdminAction::ListPluginRunsAction
+                | AdminAction::ReplayPluginRunAction
+                | AdminAction::GetPluginDelegationAction
+                | AdminAction::CreatePluginDelegationAction
+                | AdminAction::RevokePluginDelegationAction
                 | AdminAction::RunTableMaintenanceAction
                 | AdminAction::GetTableMetadataLocationAction
                 | AdminAction::SetTableMetadataLocationAction
@@ -913,6 +940,35 @@ mod tests {
         assert_eq!(arr[0].as_str().unwrap(), "s3:*");
     }
 
+    #[test]
+    fn plugin_delegation_actions_are_explicit_and_unscoped() {
+        for (name, expected) in [
+            ("admin:GetPluginDelegation", AdminAction::GetPluginDelegationAction),
+            ("admin:CreatePluginDelegation", AdminAction::CreatePluginDelegationAction),
+            ("admin:RevokePluginDelegation", AdminAction::RevokePluginDelegationAction),
+            ("admin:SubmitPluginJob", AdminAction::SubmitPluginJobAction),
+            ("admin:GetPluginRun", AdminAction::GetPluginRunAction),
+            ("admin:CancelPluginRun", AdminAction::CancelPluginRunAction),
+            ("admin:ListPluginRuns", AdminAction::ListPluginRunsAction),
+            ("admin:ReplayPluginRun", AdminAction::ReplayPluginRunAction),
+        ] {
+            assert_eq!(AdminAction::try_from(name).unwrap(), expected);
+            assert!(expected.is_valid());
+            assert!(!expected.is_table_resource_scoped());
+        }
+        assert_eq!(
+            AdminAction::try_from("admin:IssuePluginAttempt").unwrap(),
+            AdminAction::IssuePluginAttemptAction
+        );
+        assert!(AdminAction::IssuePluginAttemptAction.is_valid());
+        assert!(AdminAction::SubmitPluginJobAction.is_valid());
+        assert!(AdminAction::GetPluginRunAction.is_valid());
+        assert!(AdminAction::CancelPluginRunAction.is_valid());
+        assert!(AdminAction::ListPluginRunsAction.is_valid());
+        assert!(AdminAction::ReplayPluginRunAction.is_valid());
+        assert!(AdminAction::try_from("admin:IssuePluginCredentials").is_err());
+    }
+
     #[test]
     fn test_get_replication_metrics_admin_action_is_valid() {
         assert!(AdminAction::GetReplicationMetricsAction.is_valid());

crates/scanner/src/error.rs

@@ -53,6 +53,10 @@ pub enum ScannerError {
     #[error("Remote namespace scanner disk is already active")]
     RemoteDiskBusy,
 
+    /// Node resource pressure deferred work before its first storage poll.
+    #[error("Scanner node resource admission deferred")]
+    ResourceDeferred,
+
     /// Partial data usage cache produced before the scanner stopped.
     #[error("Scanner stopped with partial data usage cache")]
     PartialCache(Box<DataUsageCache>),

crates/scanner/src/lib.rs

@@ -50,9 +50,9 @@ use storage_api::owner::{
 };
 #[cfg(test)]
 use storage_api::owner::{
-    EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, EcstoreEndpointServerPools, EcstoreEndpoints, EcstoreInstanceContext,
-    EcstorePoolEndpoints, ecstore_config_init, ecstore_init_bucket_metadata_sys, ecstore_init_local_disks_with_instance_ctx,
-    ecstore_new_disk,
+    EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, EcstoreEndpointServerPools, EcstoreEndpoints, EcstoreExpiryState,
+    EcstoreInstanceContext, EcstorePoolEndpoints, ecstore_config_init, ecstore_init_bucket_metadata_sys,
+    ecstore_init_local_disks_with_instance_ctx, ecstore_new_disk,
 };
 use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
@@ -123,7 +123,7 @@ pub use storage_api::scan::{
     SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
     SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES,
 };
-pub use workload_admission::set_scanner_workload_admission_snapshot_provider;
+pub use workload_admission::{set_scanner_node_admission, set_scanner_workload_admission_snapshot_provider};
 
 static SCANNER_ACTIVE_WORK_UNITS: AtomicU64 = AtomicU64::new(0);
 static SCANNER_RUNTIME_INSTANCES: AtomicU64 = AtomicU64::new(0);
@@ -678,8 +678,37 @@ fn reset_tier_name_cache_for_test() {
     TIER_REGISTRY_GENERATION.store(0, Ordering::Relaxed);
 }
 
-pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo) {
-    ecstore_expiry_state_handle().write().await.enqueue_free_version(oi);
+// Only a task running an isolated unit test can override these handles.
+#[cfg(test)]
+struct ExpiryLockTestContext {
+    state: Arc<tokio::sync::RwLock<EcstoreExpiryState>>,
+    store: Option<Arc<ECStore>>,
+    completed_incarnation_reads: Arc<std::sync::atomic::AtomicUsize>,
+}
+
+#[cfg(test)]
+tokio::task_local! {
+    static EXPIRY_LOCK_TEST_CONTEXT: ExpiryLockTestContext;
+}
+
+pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo, cancellation: Option<&CancellationToken>) {
+    #[cfg(test)]
+    let state = EXPIRY_LOCK_TEST_CONTEXT
+        .try_with(|context| context.state.clone())
+        .unwrap_or_else(|_| ecstore_expiry_state_handle());
+    #[cfg(not(test))]
+    let state = ecstore_expiry_state_handle();
+    let state = match cancellation {
+        Some(cancel) => tokio::select! {
+            biased;
+            _ = cancel.cancelled() => return,
+            state = state.write() => state,
+        },
+        None => state.write().await,
+    };
+    if !cancellation.is_some_and(CancellationToken::is_cancelled) {
+        state.enqueue_free_version(oi);
+    }
 }
 
 pub(crate) async fn enqueue_runtime_newer_noncurrent(
@@ -687,20 +716,50 @@ pub(crate) async fn enqueue_runtime_newer_noncurrent(
     to_delete_objs: Vec<ObjectToDelete>,
     event: Event,
     src: &LcEventSrc,
+    cancellation: Option<&CancellationToken>,
 ) -> bool {
-    let Some(store) = ecstore_resolve_object_store_handle() else {
+    if cancellation.is_some_and(CancellationToken::is_cancelled) {
+        return false;
+    }
+    #[cfg(test)]
+    let store = EXPIRY_LOCK_TEST_CONTEXT
+        .try_with(|context| context.store.clone())
+        .unwrap_or_else(|_| ecstore_resolve_object_store_handle());
+    #[cfg(not(test))]
+    let store = ecstore_resolve_object_store_handle();
+    let Some(store) = store else {
         return false;
     };
     let Ok(bucket_incarnation_id) = store.bucket_incarnation_id(bucket).await else {
         return false;
     };
-    ecstore_expiry_state_handle().write().await.enqueue_by_newer_noncurrent(
-        bucket,
-        to_delete_objs,
-        event,
-        src,
-        bucket_incarnation_id,
-    )
+    #[cfg(test)]
+    let _ = EXPIRY_LOCK_TEST_CONTEXT.try_with(|context| {
+        context
+            .completed_incarnation_reads
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+    });
+    if cancellation.is_some_and(CancellationToken::is_cancelled) {
+        return false;
+    }
+    #[cfg(test)]
+    let state = EXPIRY_LOCK_TEST_CONTEXT
+        .try_with(|context| context.state.clone())
+        .unwrap_or_else(|_| ecstore_expiry_state_handle());
+    #[cfg(not(test))]
+    let state = ecstore_expiry_state_handle();
+    let mut state = match cancellation {
+        Some(cancel) => tokio::select! {
+            biased;
+            _ = cancel.cancelled() => return false,
+            state = state.write() => state,
+        },
+        None => state.write().await,
+    };
+    if cancellation.is_some_and(CancellationToken::is_cancelled) {
+        return false;
+    }
+    state.enqueue_by_newer_noncurrent(bucket, to_delete_objs, event, src, bucket_incarnation_id)
 }
 
 pub(crate) async fn queue_replication_heal(

crates/scanner/src/remote_scanner/stream.rs

@@ -705,7 +705,7 @@ async fn scan_and_persist_local_bucket(
         RemoteScannerServerError::worker("remote namespace scanner cache publication is blocked by data movement")
     })?;
     let cache_name = path_join_buf(&[&bucket, DATA_USAGE_CACHE_NAME]);
-    let guard = acquire_scanner_cache_locks(set.as_ref(), &cache_name, source)
+    let mut guard = acquire_scanner_cache_locks(set.as_ref(), &cache_name, source)
         .await
         .map_err(|err| {
             if err.is_contention() {
@@ -791,19 +791,23 @@ async fn scan_and_persist_local_bucket(
 
     let set_disks = scanner_set_disk_inventory(set.as_ref()).await;
     let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel::<DataUsageCache>(1);
+    let (node_tx, mut node_rx) = crate::workload_admission::scanner_node_request_channel();
     let scan_ctx = ctx.child_token();
-    let scan = ScannerIODisk::nsscanner_disk(
-        disk.clone(),
-        scan_ctx.clone(),
-        budget,
-        set_disks,
-        cache,
-        None,
-        ScannerDiskScanOptions {
-            scan_mode,
-            prefix_scan_scope: None,
-            checkpoint_tx: Some(checkpoint_tx),
-        },
+    let scan = crate::workload_admission::with_scanner_node_requests(
+        node_tx,
+        ScannerIODisk::nsscanner_disk(
+            disk.clone(),
+            scan_ctx.clone(),
+            budget,
+            set_disks,
+            cache,
+            None,
+            ScannerDiskScanOptions {
+                scan_mode,
+                prefix_scan_scope: None,
+                checkpoint_tx: Some(checkpoint_tx),
+            },
+        ),
     );
     tokio::pin!(scan);
     let fence_watch = watch_remote_scanner_request_fence(next_cycle, leader_epoch, store.clone(), NS_SCANNER_FENCE_POLL_INTERVAL);
@@ -813,6 +817,27 @@ async fn scan_and_persist_local_bucket(
     let mut checkpoint_channel_closed = false;
     let outcome = loop {
         tokio::select! {
+            request = async {
+                match node_rx.as_mut() {
+                    Some(receiver) => receiver.recv().await,
+                    None => std::future::pending().await,
+                }
+            } => {
+                let Some(request) = request else {
+                    node_rx = None;
+                    continue;
+                };
+                if let Err(error) = wait_for_remote_node_resumption(
+                    guard.resume_node_operation(set.clone(), &cache_name, source, &revisions, request),
+                    fence_watch.as_mut(),
+                ).await {
+                    scan_ctx.cancel();
+                    let _ = tokio::time::timeout(NS_SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, scan.as_mut()).await;
+                    return Err(RemoteScannerServerError::retry_bucket(format!(
+                        "remote namespace scanner could not resume node admission: {error}",
+                    )));
+                }
+            }
             result = &mut scan => {
                 match result {
                     Ok(outcome) => break outcome,
@@ -1121,6 +1146,22 @@ pub(crate) async fn scan_remote_bucket(
     result
 }
 
+/// Resource waits must keep checking the persisted leader fence.
+async fn wait_for_remote_node_resumption<F, G>(resume: F, fence_watch: Pin<&mut G>) -> Result<(), StorageError>
+where
+    F: Future<Output = Result<(), StorageError>>,
+    G: Future<Output = Result<(), ScannerError>>,
+{
+    tokio::select! {
+        biased;
+        result = fence_watch => Err(match result {
+            Ok(()) => StorageError::other("remote namespace scanner fence watcher stopped unexpectedly"),
+            Err(error) => StorageError::other(error),
+        }),
+        result = resume => result,
+    }
+}
+
 fn finish_remote_scanner_stream(
     result: RemoteScannerStreamResult<RemoteScannerOutcome>,
     budget: &ScannerCycleBudget,

crates/scanner/src/remote_scanner/stream/tests.rs

@@ -1006,6 +1006,58 @@ async fn fence_watcher_stops_work_after_persisted_leader_changes() {
     assert!(err.to_string().contains("requested_epoch=9"));
 }
 
+#[tokio::test]
+async fn node_admission_wait_does_not_hide_replacement_leader() {
+    use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
+    use std::num::NonZeroU16;
+
+    let store = Arc::new(MutableCycleStateStore {
+        state: RwLock::new(crate::scanner::encode_scanner_cycle_fence_for_test(7, 9)),
+    });
+    let cache = Mutex::new(None);
+    let refresh = AsyncMutex::new(());
+    let watcher = watch_remote_scanner_request_fence_with_cache(
+        7,
+        9,
+        store.clone(),
+        Duration::from_millis(5),
+        &cache,
+        &refresh,
+        Duration::from_millis(5),
+    );
+    tokio::pin!(watcher);
+    let gate = NodeAdmission::new(NonZeroU16::new(1).expect("one shared node unit"));
+    let storage_polled = AtomicBool::new(false);
+    {
+        let resume = async {
+            let _permit = gate.acquire().await;
+            storage_polled.store(true, Ordering::Relaxed);
+            Ok(())
+        };
+        let waiting = wait_for_remote_node_resumption(resume, watcher.as_mut());
+        tokio::pin!(waiting);
+        assert!(futures::poll!(&mut waiting).is_pending(), "no resource sample grants a node unit");
+        *store.state.write().expect("persist replacement leader") = crate::scanner::encode_scanner_cycle_fence_for_test(7, 10);
+        let error = tokio::time::timeout(Duration::from_millis(100), waiting)
+            .await
+            .expect("resource pause must not hide a changed leader")
+            .expect_err("old leader cannot resume storage work");
+        assert!(error.to_string().contains("requested_epoch=9"));
+    }
+    assert!(!storage_polled.load(Ordering::Relaxed));
+    assert_eq!(gate.snapshot().active, 0);
+    gate.update(
+        &mut FeedbackController::new(),
+        Some(FeedbackObservation {
+            generation: 1,
+            max_concurrency: 1,
+            pressure: ResourcePressure::Unknown,
+            completed_units: None,
+        }),
+    );
+    assert!(gate.try_acquire().is_some(), "failed fencing drops the old node queue ticket");
+}
+
 #[tokio::test]
 async fn concurrent_cycle_cache_misses_share_one_backend_read() {
     let request = RemoteScannerRequest(test_request(Uuid::new_v4()));

crates/scanner/src/scanner_folder.rs

@@ -43,6 +43,7 @@ use crate::storage_api::owner::{
 };
 #[cfg(test)]
 use crate::storage_api::owner::{EcstoreExpirationStatus as ExpirationStatus, EcstoreLifecycleRule as LifecycleRule};
+use crate::workload_admission::run_scanner_node_operation;
 use metrics::{counter, describe_counter};
 use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count};
 use rustfs_filemeta::{
@@ -795,6 +796,7 @@ pub struct FolderScanner {
     resolved_failed_object_paths: HashSet<String>,
 
     sleeper: DynamicSleeper,
+    node_admission: Option<Arc<rustfs_concurrency::adaptive::NodeAdmission>>,
     // should_heal: Arc<dyn Fn() -> bool + Send + Sync>,
     disks: Vec<Arc<Disk>>,
     disks_quorum: usize,
@@ -828,6 +830,8 @@ pub struct FolderScanner {
     pending_size_reconciliation_truncated: bool,
     #[cfg(test)]
     list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
+    #[cfg(test)]
+    defer_abandoned_entries_until_finished: bool,
 }
 
 struct RawEnumerationProgress {
@@ -998,12 +1002,13 @@ async fn next_scanner_dir_entry(
     ctx: &CancellationToken,
     sleeper: &DynamicSleeper,
     entries: &mut ScannerDirectory,
+    admission: Option<&Arc<rustfs_concurrency::adaptive::NodeAdmission>>,
 ) -> Result<Option<tokio::fs::DirEntry>, ScannerError> {
     if entries.batch_entries == SCANNER_DIRECTORY_BATCH_ENTRIES {
         wait_for_scanner_io(ctx, tokio::task::yield_now()).await?;
         entries.batch_entries = 0;
     }
-    let entry = wait_for_scanner_io(ctx, entries.next_entry()).await??;
+    let entry = wait_for_scanner_io(ctx, run_scanner_node_operation(admission, ctx, entries.next_entry())).await???;
     if entry.is_some() {
         // Tokio prefetches bounded chunks. Pacing consumption also paces the
         // next refill, without charging an extra entry for EOF.
@@ -1923,7 +1928,12 @@ impl FolderScanner {
             // directory that vanished during iteration instead of failing the
             // whole parent subtree. The bucket root (no parent) still propagates
             // NotFound so namespace removal is detected by the caller.
-            let mut dir_reader = match wait_for_scanner_io(&ctx, ScannerDirectory::open(&dir_path)).await? {
+            let mut dir_reader = match wait_for_scanner_io(
+                &ctx,
+                run_scanner_node_operation(self.node_admission.as_ref(), &ctx, ScannerDirectory::open(&dir_path)),
+            )
+            .await??
+            {
                 Ok(dir_reader) => Some(dir_reader),
                 Err(e)
                     if folder.parent.is_some()
@@ -1961,7 +1971,7 @@ impl FolderScanner {
                 // Hold at most one metadata entry until a candidate reaches
                 // EOF: a later part must not leave object accounting or ILM work.
                 let (entry, replayed_metadata) = if let Some(reader) = dir_reader.as_mut() {
-                    let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader).await {
+                    let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader, self.node_admission.as_ref()).await {
                         Ok(Some(entry)) => entry,
                         Ok(None) => {
                             raw_enumeration_complete = true;
@@ -2072,7 +2082,12 @@ impl FolderScanner {
 
                 // Ignore entries that disappeared during traversal or hit symlink
                 // loops, but propagate other walk errors.
-                let mut entry_type = match wait_for_scanner_io(&ctx, entry.file_type()).await? {
+                let mut entry_type = match wait_for_scanner_io(
+                    &ctx,
+                    run_scanner_node_operation(self.node_admission.as_ref(), &ctx, entry.file_type()),
+                )
+                .await??
+                {
                     Ok(entry_type) => entry_type,
                     Err(e) if e.kind() == ErrorKind::NotFound => {
                         debug!(
@@ -2123,7 +2138,12 @@ impl FolderScanner {
                 let mut entry_metadata_len = None;
                 if entry_type.is_symlink() {
                     wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
-                    let metadata = match wait_for_scanner_io(&ctx, tokio::fs::metadata(&file_path)).await? {
+                    let metadata = match wait_for_scanner_io(
+                        &ctx,
+                        run_scanner_node_operation(self.node_admission.as_ref(), &ctx, tokio::fs::metadata(&file_path)),
+                    )
+                    .await??
+                    {
                         Ok(metadata) => metadata,
                         Err(e) if e.kind() == ErrorKind::NotFound => {
                             debug!(
@@ -2274,22 +2294,30 @@ impl FolderScanner {
                         len
                     } else {
                         wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
-                        wait_for_scanner_io(&ctx, tokio::fs::metadata(&item.path))
-                            .await?
-                            .map(|metadata| metadata.len())
-                            .unwrap_or(0)
+                        wait_for_scanner_io(
+                            &ctx,
+                            run_scanner_node_operation(self.node_admission.as_ref(), &ctx, tokio::fs::metadata(&item.path)),
+                        )
+                        .await??
+                        .map(|metadata| metadata.len())
+                        .unwrap_or(0)
                     }
                 } else {
                     0
                 };
                 wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, budgeted_bytes)).await?;
 
-                self.object_metadata_visits.increment(1);
-                let sz = match self
-                    .local_disk
-                    .get_size_with_tier_names(item.clone(), &self.tier_registry.names)
-                    .await
-                {
+                let size_result = run_scanner_node_operation(self.node_admission.as_ref(), &ctx, async {
+                    self.object_metadata_visits.increment(1);
+                    self.local_disk
+                        .get_size_with_tier_names(item.clone(), &self.tier_registry.names)
+                        .await
+                })
+                .await?;
+                if self.node_admission.is_some() && ctx.is_cancelled() {
+                    return Err(ScannerError::Other("Operation cancelled".to_string()));
+                }
+                let sz = match size_result {
                     Ok(sz) => sz,
                     Err(e) => {
                         let failure_action = classify_get_size_failure(&item, &e);
@@ -2606,7 +2634,7 @@ impl FolderScanner {
                     {
                         Ok(outcome) => outcome,
                         Err(e) => {
-                            if ctx.is_cancelled() {
+                            if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
                                 self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
                                     .await;
                                 return Err(e);
@@ -2847,19 +2875,23 @@ impl FolderScanner {
                 let mut seen_truncated_objects: HashSet<String> = HashSet::new();
 
                 loop {
-                    if agreed_closed && partial_closed && finished_closed {
+                    if child_ctx.is_cancelled() || (agreed_closed && partial_closed && finished_closed) {
                         break;
                     }
+                    #[cfg(test)]
+                    let defer_entries = self.defer_abandoned_entries_until_finished && !finished_closed;
+                    #[cfg(not(test))]
+                    let defer_entries = false;
 
                     select! {
-                        entry_name = agreed_rx.recv(), if !agreed_closed => {
+                        entry_name = agreed_rx.recv(), if !agreed_closed && !defer_entries => {
                             let Some(entry_name) = entry_name else {
                                 agreed_closed = true;
                                 continue;
                             };
                             (self.update_current_path)(&entry_name).await;
                         }
-                        entries = partial_rx.recv(), if !partial_closed => {
+                        entries = partial_rx.recv(), if !partial_closed && !defer_entries => {
                             let Some(entries) = entries else {
                                 partial_closed = true;
                                 continue;
@@ -2985,6 +3017,7 @@ impl FolderScanner {
                                 finished_closed = true;
                                 continue;
                             };
+                            finished_closed = true;
                             if disk_errors_are_only_missing_paths(&errs) {
                                 debug!(
                                     target: "rustfs::scanner::folder",
@@ -3006,7 +3039,9 @@ impl FolderScanner {
                                     "Scanner list_path_raw finished with disk errors"
                                 );
                             }
-                            child_ctx.cancel();
+                            // The producer awaited callback sends, not their
+                            // consumption. Drain the remaining candidate and
+                            // agreed frames before closing this child walk.
                         }
                         _ = child_ctx.cancelled() => {
                             break;
@@ -3024,7 +3059,10 @@ impl FolderScanner {
                     if into.compacted {
                         // In compacted mode child totals are accumulated directly into the parent entry.
                         let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
-                        fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
+                        fut.await.map_err(|error| match error {
+                            ScannerError::ResourceDeferred => ScannerError::ResourceDeferred,
+                            other => ScannerError::Other(other.to_string()),
+                        })?;
                         self.send_update_for_entry(&this_hash, &folder.parent, into).await;
                         tokio::task::yield_now().await;
                     } else {
@@ -3043,7 +3081,7 @@ impl FolderScanner {
                             )
                             .await
                         {
-                            if ctx.is_cancelled() {
+                            if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
                                 self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
                                     .await;
                                 return Err(e);
@@ -3217,6 +3255,7 @@ pub(crate) async fn scan_data_folder_scoped(
         failed_object_paths_seen: HashSet::new(),
         resolved_failed_object_paths: HashSet::new(),
         sleeper,
+        node_admission: crate::workload_admission::scanner_node_admission(),
         disks,
         disks_quorum,
         usage_updates,
@@ -3244,6 +3283,8 @@ pub(crate) async fn scan_data_folder_scoped(
         pending_size_reconciliation_truncated: false,
         #[cfg(test)]
         list_path_raw_options_observer: None,
+        #[cfg(test)]
+        defer_abandoned_entries_until_finished: false,
     };
 
     let now = FolderScanner::now_secs();
@@ -3333,7 +3374,7 @@ pub(crate) async fn scan_data_folder_scoped(
             }
         }
         Err(e) => {
-            if ctx.is_cancelled() {
+            if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
                 let root_hash = hash_path(&bucket_name);
                 let root_has_progress = data_usage_root_has_progress(&root);
                 let pending_heals_changed = scanner.pending_heals_changed;
@@ -3408,6 +3449,19 @@ pub(crate) async fn scan_data_folder_scoped(
                 close_disk_guard.close().await;
                 return Err(ScannerError::NamespaceNotFoundCache(Box::new(partial_cache)));
             }
+            if matches!(&e, ScannerError::ResourceDeferred) {
+                // Admission pressure is not disk failure or an empty census.
+                // Keep the previous tree when no new useful progress exists.
+                let mut partial_cache = scanner.old_cache;
+                if partial_cache.cache.is_empty() {
+                    partial_cache
+                        .cache
+                        .insert(hash_path(&bucket_name).0, DataUsageEntry::default());
+                }
+                partial_cache.info.snapshot_complete = false;
+                close_disk_guard.close().await;
+                return Err(ScannerError::PartialCache(Box::new(partial_cache)));
+            }
             close_disk_guard.close().await;
             // No useful information, return original cache
             Err(e)
@@ -3427,3 +3481,15 @@ pub(crate) use tests::checkpoint_fixture::run_checkpoint_fixture;
 
 #[cfg(test)]
 mod tests;
+
+#[cfg(test)]
+tokio::task_local! {
+    pub(crate) static TEST_SIZE_DISPATCH_BOUNDARIES: tokio::sync::mpsc::UnboundedSender<&'static str>;
+}
+
+#[cfg(test)]
+pub(crate) fn observe_test_size_dispatch_boundary(stage: &'static str) -> bool {
+    TEST_SIZE_DISPATCH_BOUNDARIES
+        .try_with(|sender| sender.send(stage).is_ok())
+        .unwrap_or(false)
+}

crates/scanner/src/scanner_folder/item_actions.rs

@@ -628,6 +628,22 @@ impl ScannerItem {
         tier_names: &[String],
         size_summary: &mut SizeSummary,
     ) {
+        self.apply_actions_cancellable(object_infos, lock_retention, versioning_config, tier_names, size_summary, None)
+            .await;
+    }
+
+    pub(crate) async fn apply_actions_cancellable(
+        &mut self,
+        object_infos: Vec<ObjectInfo>,
+        lock_retention: Option<Arc<ObjectLockConfiguration>>,
+        versioning_config: VersioningConfiguration,
+        tier_names: &[String],
+        size_summary: &mut SizeSummary,
+        cancellation: Option<&CancellationToken>,
+    ) {
+        if cancellation.is_some_and(CancellationToken::is_cancelled) {
+            return;
+        }
         let object_path = self.object_path();
         if object_infos.is_empty() {
             debug!(
@@ -671,6 +687,9 @@ impl ScannerItem {
         let Some(lifecycle) = self.lifecycle.clone() else {
             let mut cumulative_size: i64 = 0;
             for (oi, resolved_size) in object_infos.iter().zip(resolved_sizes.iter()) {
+                if cancellation.is_some_and(CancellationToken::is_cancelled) {
+                    return;
+                }
                 let accounting_size = match resolved_size {
                     SizeResolution::Known { logical, .. } => *logical,
                     // A valid compressed legacy sentinel has no logical size,
@@ -678,7 +697,7 @@ impl ScannerItem {
                     // physical size is only an input to those operations; it
                     // is not folded into the logical total below.
                     SizeResolution::Unknown { physical, .. } => {
-                        self.heal_actions(oi, *physical, size_summary).await;
+                        self.heal_actions(oi, *physical, size_summary, cancellation).await;
                         size_summary.actions_accounting_unknown(oi);
                         continue;
                     }
@@ -688,13 +707,16 @@ impl ScannerItem {
                     }
                 };
 
-                let size = self.heal_actions(oi, accounting_size, size_summary).await;
+                let size = self.heal_actions(oi, accounting_size, size_summary, cancellation).await;
 
                 size_summary.actions_accounting(oi, size, accounting_size);
 
                 cumulative_size = cumulative_size.saturating_add(size);
             }
 
+            if cancellation.is_some_and(CancellationToken::is_cancelled) {
+                return;
+            }
             self.alert_excessive_versions(object_infos.len(), cumulative_size);
 
             debug!(
@@ -736,6 +758,10 @@ impl ScannerItem {
             }
         };
 
+        if cancellation.is_some_and(CancellationToken::is_cancelled) {
+            return;
+        }
+
         // Every version handed to the evaluator counts as an ILM-checked
         // version, whether or not a rule matched (NoneAction included). This is
         // reached only for buckets with lifecycle rules; it feeds
@@ -754,6 +780,9 @@ impl ScannerItem {
             .all(|candidate| Self::tier_is_known(candidate, tier_names));
         'eventLoop: {
             for (i, event) in events.iter().enumerate() {
+                if cancellation.is_some_and(CancellationToken::is_cancelled) {
+                    return;
+                }
                 let oi = &object_infos[i];
                 let known_size = resolved_sizes[i].known_size();
                 if has_corrupt_size
@@ -772,7 +801,7 @@ impl ScannerItem {
                     // accounting. Size-filtered or deferred events remain
                     // pending, so retain the version-only physical counters.
                     if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
-                        self.heal_actions(oi, *physical, size_summary).await;
+                        self.heal_actions(oi, *physical, size_summary, cancellation).await;
                         size_summary.actions_accounting_unknown(oi);
                     }
                     continue;
@@ -829,7 +858,7 @@ impl ScannerItem {
                                     noncurrent_unknown.push(oi);
                                 } else {
                                     if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
-                                        self.heal_actions(oi, *physical, size_summary).await;
+                                        self.heal_actions(oi, *physical, size_summary, cancellation).await;
                                     }
                                     size_summary.actions_accounting_unknown(oi);
                                 }
@@ -846,7 +875,7 @@ impl ScannerItem {
                             }
                             IlmAction::NoneAction | IlmAction::ActionCount => {
                                 if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
-                                    self.heal_actions(oi, *physical, size_summary).await;
+                                    self.heal_actions(oi, *physical, size_summary, cancellation).await;
                                 }
                                 size_summary.actions_accounting_unknown(oi);
                             }
@@ -864,7 +893,7 @@ impl ScannerItem {
                 // side-effecting tier actions fail closed until the registry
                 // recognizes the source again.
                 if Self::action_blocked_by_unknown_tier(event.action, oi, all_versions_known, tier_names, &event.storage_class) {
-                    size = self.heal_actions(oi, actual_size, size_summary).await;
+                    size = self.heal_actions(oi, actual_size, size_summary, cancellation).await;
                     size_summary.actions_accounting(oi, size, actual_size);
                     cumulative_size += size;
                     continue;
@@ -973,7 +1002,7 @@ impl ScannerItem {
                     }
 
                     IlmAction::NoneAction | IlmAction::ActionCount => {
-                        size = self.heal_actions(oi, actual_size, size_summary).await;
+                        size = self.heal_actions(oi, actual_size, size_summary, cancellation).await;
                     }
                 }
 
@@ -984,6 +1013,9 @@ impl ScannerItem {
             }
         }
 
+        if cancellation.is_some_and(CancellationToken::is_cancelled) {
+            return;
+        }
         if !to_delete_objs.is_empty()
             && let Some(event) = noncurrent_events.first().cloned()
         {
@@ -991,7 +1023,8 @@ impl ScannerItem {
             let count = u64::try_from(to_delete_objs.len()).unwrap_or(u64::MAX);
             let done_ilm = Metrics::time_ilm(action);
             let trace_started_at = trace_start_instant();
-            let queued = enqueue_runtime_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner).await;
+            let queued =
+                enqueue_runtime_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner, cancellation).await;
             if let Some(trace_started_at) = trace_started_at {
                 let state = if queued { "queued" } else { "not_queued" };
                 trace_emit(|| {
@@ -1022,15 +1055,28 @@ impl ScannerItem {
                 }
             }
         }
-        self.alert_excessive_versions(remaining_versions, cumulative_size);
+        if !cancellation.is_some_and(CancellationToken::is_cancelled) {
+            self.alert_excessive_versions(remaining_versions, cumulative_size);
+        }
     }
 
-    pub(super) async fn heal_actions(&mut self, oi: &ObjectInfo, actual_size: i64, size_summary: &mut SizeSummary) -> i64 {
+    pub(super) async fn heal_actions(
+        &mut self,
+        oi: &ObjectInfo,
+        actual_size: i64,
+        size_summary: &mut SizeSummary,
+        cancellation: Option<&CancellationToken>,
+    ) -> i64 {
+        if cancellation.is_some_and(CancellationToken::is_cancelled) {
+            return actual_size;
+        }
         if self.heal_enabled {
             self.enqueue_heal(oi).await;
         }
 
-        self.heal_replication(oi, size_summary).await;
+        if !cancellation.is_some_and(CancellationToken::is_cancelled) {
+            self.heal_replication(oi, size_summary).await;
+        }
 
         actual_size
     }

crates/scanner/src/scanner_folder/tests.rs

@@ -30,7 +30,9 @@ use std::sync::Mutex;
 pub(super) mod checkpoint_fixture;
 pub(super) mod enumeration_restart;
 mod incremental_enumeration;
+mod listing_completion;
 mod metadata_visits;
+mod node_admission;
 mod usage_updates;
 
 /// Reset the process-global alert cooldown map; test-only.
@@ -364,6 +366,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         failed_object_paths_seen: HashSet::new(),
         resolved_failed_object_paths: HashSet::new(),
         sleeper: SCANNER_SLEEPER.clone(),
+        node_admission: None,
         disks: Vec::new(),
         disks_quorum: 0,
         usage_updates: None,
@@ -394,6 +397,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         pending_size_reconciliation_scopes: HashSet::new(),
         pending_size_reconciliation_truncated: false,
         list_path_raw_options_observer: None,
+        defer_abandoned_entries_until_finished: false,
     };
 
     (scanner, temp_dir)
@@ -2262,7 +2266,7 @@ async fn test_heal_actions_returns_actual_size_without_inline_heal() {
     };
     let mut size_summary = SizeSummary::default();
 
-    let size = item.heal_actions(&object_info, 123, &mut size_summary).await;
+    let size = item.heal_actions(&object_info, 123, &mut size_summary, None).await;
     assert_eq!(size, 123);
 }
 

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

@@ -0,0 +1,104 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+
+#[tokio::test]
+#[serial]
+async fn abandoned_listing_drains_candidate_buffered_before_finished() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let bucket = "bucket";
+    let object = "abandoned/object";
+    tokio::fs::create_dir_all(root.join(bucket)).await.expect("local bucket");
+    let mut disks = vec![scanner.local_disk.clone()];
+    let mut original_metadata = Vec::new();
+    for (index, version) in [(2, 0x1935), (3, 0x1935), (4, 0x1934)] {
+        let disk_root = root.join(format!("disk-{index}"));
+        let metadata = metadata_for_object_version(bucket, object, Some(Uuid::from_u128(version)));
+        write_test_object_metadata_bytes(&disk_root, bucket, object, &metadata).await;
+        original_metadata.push((disk_root.join(bucket).join(object).join(STORAGE_FORMAT_FILE), metadata));
+        let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("listing fixture endpoint");
+        disks.push(
+            new_disk(
+                &endpoint,
+                &DiskOption {
+                    cleanup: false,
+                    health_check: false,
+                },
+            )
+            .await
+            .expect("listing fixture disk"),
+        );
+    }
+    for cache in [&mut scanner.old_cache, &mut scanner.new_cache] {
+        cache.info.name = bucket.to_owned();
+    }
+    scanner.old_cache.replace(
+        "bucket/abandoned/object",
+        bucket,
+        DataUsageEntry {
+            objects: 1,
+            ..Default::default()
+        },
+    );
+    scanner.heal_object_select = 1;
+    scanner.disks = disks;
+    scanner.disks_quorum = 2;
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    // Force the valid select ordering that receives terminal errors before the
+    // final candidate. One candidate frame fits in the bounded partial channel.
+    scanner.defer_abandoned_entries_until_finished = true;
+    let (_commands, mut heal_rx) = rustfs_heal_contracts::heal_channel::init_heal_channels().expect("isolated heal channel");
+    let (captured_tx, mut captured_rx) = mpsc::unbounded_channel();
+    let responder = tokio::spawn(async move {
+        while let Some(command) = heal_rx.recv().await {
+            let rustfs_heal_contracts::heal_channel::HealReceiptCommand { request, response_tx } = command;
+            captured_tx
+                .send((request.object_prefix.clone(), request.object_version_id.clone()))
+                .expect("candidate capture remains open");
+            let _ = response_tx.send(Ok(rustfs_heal_contracts::heal_channel::HealAdmissionReceipt {
+                result: HealAdmissionResult::Accepted,
+                task_id: request.id,
+            }));
+        }
+    });
+    let result = tokio::time::timeout(
+        Duration::from_secs(2),
+        scanner.scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: bucket.to_owned(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut DataUsageEntry::default(),
+        ),
+    )
+    .await;
+    responder.abort();
+    result
+        .expect("listing completion must close every callback channel")
+        .expect("complete folder scan");
+    let mut candidates = HashSet::new();
+    while let Ok((object, version)) = captured_rx.try_recv() {
+        if let (Some(object), Some(version)) = (object, version) {
+            candidates.insert((object, version));
+        }
+    }
+    let expected = [0x1934, 0x1935]
+        .into_iter()
+        // Raw base-object listings include the trailing separator. Preserve
+        // the producer's candidate identity when draining its callback frame.
+        .map(|version| (format!("{object}/"), Uuid::from_u128(version).to_string()))
+        .collect();
+    assert_eq!(
+        candidates, expected,
+        "terminal disk errors must not discard buffered exact-version candidates"
+    );
+    for (path, metadata) in original_metadata {
+        assert_eq!(tokio::fs::read(path).await.expect("retained metadata"), metadata);
+    }
+}

crates/scanner/src/scanner_io/cache.rs

@@ -172,12 +172,55 @@ fn record_checkpoint_save_failure(cache_name: &str, checkpoint: &DataUsageCache,
 
 #[derive(Debug)]
 pub(crate) struct ScannerCacheLockGuards {
-    scoped: NamespaceLockGuard,
+    scoped: Option<NamespaceLockGuard>,
 }
 
 impl ScannerCacheLockGuards {
     pub(crate) fn is_lock_lost(&self) -> bool {
-        self.scoped.is_lock_lost()
+        self.scoped.as_ref().is_none_or(NamespaceLockGuard::is_lock_lost)
+    }
+
+    /// Suspend the same traversal while yielding its cache lease. Lock order on
+    /// resumption is node admission -> cache lease -> cache revision reads.
+    pub(crate) async fn resume_node_operation(
+        &mut self,
+        store: Arc<SetDisks>,
+        cache_name: &str,
+        source: DataUsageCacheSource,
+        revisions: &DataUsageCacheRevisions,
+        request: crate::workload_admission::ScannerNodeRequest,
+    ) -> Result<()> {
+        if self.is_lock_lost() {
+            return Err(Error::other("scanner cache lock was lost before node admission"));
+        }
+        if request.respond.is_closed() {
+            return Ok(());
+        }
+        let crate::workload_admission::ScannerNodeRequest { owner, respond, cancel } = request;
+        drop(self.scoped.take());
+        let resume = async {
+            let permit = owner.acquire().await;
+            *self = acquire_scanner_cache_locks(store.as_ref(), cache_name, source)
+                .await
+                .map_err(Error::other)?;
+            let current = crate::storage_api::ecstore_with_background_disk_io(
+                cancel.clone(),
+                DataUsageCache::read_revisions(store, cache_name),
+            )
+            .await?;
+            if self.is_lock_lost() || &current != revisions {
+                return Err(Error::other("scanner cache changed while waiting for node admission"));
+            }
+            // If the walk was dropped, send returns ownership and drops the
+            // permit without recording completed work.
+            let _ = respond.send(permit);
+            Ok(())
+        };
+        tokio::select! {
+            biased;
+            _ = cancel.cancelled() => Err(Error::other("scanner node admission cancelled")),
+            result = resume => result,
+        }
     }
 }
 
@@ -215,6 +258,15 @@ impl std::fmt::Display for ScannerCacheLockError {
     }
 }
 
+impl std::error::Error for ScannerCacheLockError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::Create { source, .. } => Some(source),
+            Self::Acquire { source, .. } => Some(source),
+        }
+    }
+}
+
 pub(crate) async fn acquire_scanner_cache_locks(
     store: &SetDisks,
     cache_name: &str,
@@ -237,7 +289,7 @@ pub(crate) async fn acquire_scanner_cache_locks(
             source,
         })?;
 
-    Ok(ScannerCacheLockGuards { scoped })
+    Ok(ScannerCacheLockGuards { scoped: Some(scoped) })
 }
 
 pub(super) async fn await_scanner_disk_shutdown<F>(scan: Pin<&mut F>)

crates/scanner/src/scanner_io/io_cache.rs

@@ -1090,7 +1090,7 @@ impl ScannerIOCache for SetDisks {
 
                     // Lock order: scanner leader fence -> set-scoped per-bucket cache lock ->
                     // cache object read/write.
-                    let cache_guard = match acquire_scanner_cache_locks(store_clone_clone.as_ref(), &cache_name, source).await {
+                    let mut cache_guard = match acquire_scanner_cache_locks(store_clone_clone.as_ref(), &cache_name, source).await {
                         Ok(guard) => guard,
                         Err(e) => {
                             if e.is_contention() {
@@ -1320,9 +1320,10 @@ impl ScannerIOCache for SetDisks {
                     );
 
                     let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel::<DataUsageCache>(1);
+                    let (node_tx, mut node_rx) = crate::workload_admission::scanner_node_request_channel();
 
                     let scan_ctx = ctx_clone.child_token();
-                    let scan = disk.clone().nsscanner_disk(
+                    let scan = crate::workload_admission::with_scanner_node_requests(node_tx, disk.clone().nsscanner_disk(
                         scan_ctx.clone(),
                         budget_clone.clone(),
                         set_disk_inventory_clone.as_ref().clone(),
@@ -1333,7 +1334,7 @@ impl ScannerIOCache for SetDisks {
                             prefix_scan_scope,
                             checkpoint_tx: Some(checkpoint_tx),
                         },
-                    );
+                    ));
                     tokio::pin!(scan);
                     let mut lock_watch = tokio::time::interval(SCANNER_CACHE_LOCK_POLL_INTERVAL);
                     lock_watch.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -1341,6 +1342,24 @@ impl ScannerIOCache for SetDisks {
                     let scan_result = loop {
                         tokio::select! {
                             result = &mut scan => break result,
+                            request = async {
+                                match node_rx.as_mut() {
+                                    Some(receiver) => receiver.recv().await,
+                                    None => std::future::pending().await,
+                                }
+                            } => {
+                                let Some(request) = request else {
+                                    node_rx = None;
+                                    continue;
+                                };
+                                if let Err(error) = cache_guard.resume_node_operation(
+                                    store_clone_clone.clone(), &cache_name, source, &revisions, request,
+                                ).await {
+                                    scan_ctx.cancel();
+                                    await_scanner_disk_shutdown(scan.as_mut()).await;
+                                    break Err(error);
+                                }
+                            }
                             _ = lock_watch.tick() => {
                                 if cache_guard.is_lock_lost() {
                                     scan_ctx.cancel();

crates/scanner/src/scanner_io/io_disk.rs

@@ -38,6 +38,13 @@ pub(super) fn tier_stats_template(tier_names: &[String]) -> HashMap<String, Tier
     tier_stats
 }
 
+fn ensure_scanner_node_operation_active(cancellation: Option<&CancellationToken>) -> Result<()> {
+    if cancellation.is_some_and(CancellationToken::is_cancelled) {
+        return Err(StorageError::other("Scanner operation cancelled"));
+    }
+    Ok(())
+}
+
 #[async_trait::async_trait]
 impl ScannerIODisk for Disk {
     async fn get_size(&self, item: ScannerItem) -> Result<SizeSummary> {
@@ -45,14 +52,18 @@ impl ScannerIODisk for Disk {
     }
 
     async fn get_size_with_tier_names(&self, mut item: ScannerItem, tier_names: &[String]) -> Result<SizeSummary> {
+        let cancellation = crate::workload_admission::scanner_node_operation_cancellation();
         let done_object = Metrics::time(Metric::ScanObject);
 
         if !is_xl_meta_path(&item.path) {
             return Err(StorageError::other(SCANNER_SKIP_FILE_ERROR.to_string()));
         }
 
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
         let metadata_object_path = item.object_path();
-        let data = match self.read_metadata(&item.bucket, &metadata_object_path).await {
+        let metadata_result = self.read_metadata(&item.bucket, &metadata_object_path).await;
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
+        let data = match metadata_result {
             Ok(data) => data,
             Err(e) if DiskError::is_err_object_not_found(&e) || DiskError::is_err_version_not_found(&e) => {
                 return Err(StorageError::other(SCANNER_SKIP_FILE_ERROR.to_string()));
@@ -86,7 +97,10 @@ impl ScannerIODisk for Disk {
         // (which used to query it a second time). On failure keep the
         // historical fallback: default configuration (versioned = false) plus
         // the warn that `apply_actions` used to emit.
-        let versioning_config = match BucketVersioningSys::get(&item.bucket).await {
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
+        let versioning_result = BucketVersioningSys::get(&item.bucket).await;
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
+        let versioning_config = match versioning_result {
             Ok(versioning_config) => versioning_config,
             Err(_) => {
                 warn!(
@@ -122,7 +136,9 @@ impl ScannerIODisk for Disk {
             ..Default::default()
         };
 
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
         let lock_config = object_lock_config_for_scanner_item(&item).await;
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
 
         // Count every version this object contributes to the scan, independent
         // of any lifecycle configuration, so scan-coverage metrics stay honest
@@ -130,17 +146,33 @@ impl ScannerIODisk for Disk {
         // `object_infos`.
         global_metrics().record_scanner_versions_scanned(object_infos.len() as u64);
 
-        item.apply_actions(object_infos, lock_config, versioning_config, tier_names, &mut size_summary)
-            .await;
+        #[cfg(test)]
+        let _ = crate::scanner_folder::observe_test_size_dispatch_boundary("apply_actions");
+        item.apply_actions_cancellable(
+            object_infos,
+            lock_config,
+            versioning_config,
+            tier_names,
+            &mut size_summary,
+            cancellation.as_ref(),
+        )
+        .await;
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
 
         if !free_version_infos.is_empty() {
             for oi in free_version_infos {
+                ensure_scanner_node_operation_active(cancellation.as_ref())?;
                 if ScannerItem::tier_is_known(&oi, tier_names) {
-                    enqueue_runtime_free_version(oi).await;
+                    #[cfg(test)]
+                    if crate::scanner_folder::observe_test_size_dispatch_boundary("free_version_dispatch") {
+                        continue;
+                    }
+                    enqueue_runtime_free_version(oi, cancellation.as_ref()).await;
                 }
             }
         }
 
+        ensure_scanner_node_operation_active(cancellation.as_ref())?;
         done_object();
 
         Ok(size_summary)

crates/scanner/src/scanner_io/tests.rs

@@ -46,7 +46,9 @@ use uuid::Uuid;
 
 mod checkpoint_coverage_identity;
 mod collector_lifecycle;
+mod expiry_lock;
 mod listing_window_scope;
+mod node_admission;
 mod partial_bucket_carry;
 mod per_bucket_certification;
 mod scoped_entry_fallback;

crates/scanner/src/scanner_io/tests/expiry_lock.rs

@@ -0,0 +1,183 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use crate::storage_api::owner::EcstoreExpiryState;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::task::Poll;
+use std::time::Duration;
+
+// Return Some(result) if the operation ended too early, None only if it was
+// actually polled pending after the prerequisite read had finished.
+async fn observe_queued<F: Future>(
+    mut operation: Pin<&mut F>,
+    read_complete: impl Fn() -> bool,
+    observation: Duration,
+) -> std::result::Result<Option<F::Output>, tokio::time::error::Elapsed> {
+    tokio::time::timeout(
+        observation,
+        std::future::poll_fn(|cx| match operation.as_mut().poll(cx) {
+            Poll::Ready(value) => Poll::Ready(Some(value)),
+            Poll::Pending if read_complete() => Poll::Ready(None),
+            Poll::Pending => Poll::Pending,
+        }),
+    )
+    .await
+}
+
+#[tokio::test]
+#[serial]
+async fn free_version_cancelled_while_waiting_for_expiry_write_lock_never_queues() {
+    let state = EcstoreExpiryState::new_with_unconsumed_worker_channel(4);
+    let observation = Duration::from_secs(rustfs_config::DEFAULT_DRIVE_METADATA_TIMEOUT_SECS);
+    let context = crate::ExpiryLockTestContext {
+        state: state.clone(),
+        store: None,
+        completed_incarnation_reads: Arc::new(AtomicUsize::new(0)),
+    };
+    crate::EXPIRY_LOCK_TEST_CONTEXT
+        .scope(context, async {
+            let object = crate::ScannerObjectInfo {
+                bucket: "expiry-lock-bucket".to_string(),
+                name: "free-object".to_string(),
+                ..Default::default()
+            };
+            let cancellation = CancellationToken::new();
+            let guard = state.write().await;
+            let mut pending = Box::pin(crate::enqueue_runtime_free_version(object.clone(), Some(&cancellation)));
+            let first = observe_queued(pending.as_mut(), || true, observation).await;
+            let observed_wait = matches!(&first, Ok(None));
+            let already_finished = matches!(&first, Ok(Some(())));
+            cancellation.cancel();
+            let cancelled_while_locked = if already_finished {
+                false
+            } else {
+                tokio::time::timeout(observation, pending.as_mut()).await.is_ok()
+            };
+            let pending_while_locked = guard.pending_tasks();
+            drop(guard);
+            if !already_finished && !cancelled_while_locked {
+                pending.as_mut().await;
+            }
+            drop(pending);
+            let pending_after_cancel = state.read().await.pending_tasks();
+
+            // Positive control queues a real free-version task while using the same
+            // implementation and isolated, unconsumed worker channel.
+            let mut positive = Box::pin(crate::enqueue_runtime_free_version(object, None));
+            let positive_in_time = tokio::time::timeout(observation, positive.as_mut()).await.is_ok();
+            if !positive_in_time {
+                positive.as_mut().await;
+            }
+            drop(positive);
+            let pending_after_control = state.read().await.pending_tasks();
+            assert!(observed_wait, "must poll the real enqueue while the write lock is held");
+            assert!(cancelled_while_locked, "cancellation must finish before the lock is released");
+            assert_eq!(pending_while_locked, 0);
+            assert_eq!(pending_after_cancel, 0);
+            assert!(positive_in_time, "positive control must finish within the observation limit");
+            assert_eq!(pending_after_control, 1, "positive control must queue a task");
+        })
+        .await;
+}
+
+#[tokio::test]
+#[serial]
+async fn newer_noncurrent_cancelled_after_real_read_while_waiting_for_expiry_write_lock_never_queues() {
+    let (_temp_dir, store) = setup_local_scanner_store(1, 4).await;
+    let bucket = "expiry-lock-bucket";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket");
+    let expected_incarnation = store.bucket_incarnation_id(bucket).await.expect("real incarnation metadata");
+    assert!(!expected_incarnation.is_nil());
+    let state = EcstoreExpiryState::new_with_unconsumed_worker_channel(4);
+    let reads = Arc::new(AtomicUsize::new(0));
+    let context = crate::ExpiryLockTestContext {
+        state: state.clone(),
+        store: Some(store.clone()),
+        completed_incarnation_reads: reads.clone(),
+    };
+    let observation = Duration::from_secs(rustfs_config::DEFAULT_DRIVE_METADATA_TIMEOUT_SECS);
+    crate::EXPIRY_LOCK_TEST_CONTEXT
+        .scope(context, async {
+            let versions = vec![crate::ScannerObjectToDelete {
+                object_name: "noncurrent-object".to_string(),
+                version_id: Some(Uuid::new_v4()),
+                ..Default::default()
+            }];
+            let event = crate::Event::default();
+            let src = crate::LcEventSrc::Scanner;
+            let cancellation = CancellationToken::new();
+            let guard = state.write().await;
+            let mut pending = Box::pin(crate::enqueue_runtime_newer_noncurrent(
+                bucket,
+                versions.clone(),
+                event.clone(),
+                &src,
+                Some(&cancellation),
+            ));
+            let first = observe_queued(pending.as_mut(), || reads.load(Ordering::SeqCst) == 1, observation).await;
+            let observed_wait_after_read = matches!(&first, Ok(None));
+            let mut result = match first {
+                Ok(Some(value)) => Some(value),
+                _ => None,
+            };
+            cancellation.cancel();
+            let cancelled_while_locked = if result.is_some() {
+                false
+            } else {
+                match tokio::time::timeout(observation, pending.as_mut()).await {
+                    Ok(value) => {
+                        result = Some(value);
+                        true
+                    }
+                    Err(_) => false,
+                }
+            };
+            let pending_while_locked = guard.pending_tasks();
+            drop(guard);
+            let result = match result {
+                Some(value) => value,
+                None => pending.as_mut().await,
+            };
+            drop(pending);
+            let pending_after_cancel = state.read().await.pending_tasks();
+            let mut positive = Box::pin(crate::enqueue_runtime_newer_noncurrent(bucket, versions, event, &src, None));
+            let bounded_positive = tokio::time::timeout(observation, positive.as_mut()).await;
+            let positive_in_time = bounded_positive.is_ok();
+            let positive_result = match bounded_positive {
+                Ok(value) => value,
+                Err(_) => positive.as_mut().await,
+            };
+            drop(positive);
+            let pending_after_control = state.read().await.pending_tasks();
+            assert!(
+                observed_wait_after_read,
+                "real incarnation read must finish before waiting on the held lock"
+            );
+            assert!(cancelled_while_locked, "cancellation must finish before the lock is released");
+            assert!(!result);
+            assert_eq!(pending_while_locked, 0);
+            assert_eq!(pending_after_cancel, 0);
+            assert!(positive_in_time, "positive control must finish within the observation limit");
+            assert!(positive_result, "positive control must enqueue the nonempty version batch");
+            assert_eq!(pending_after_control, 1);
+            assert_eq!(reads.load(Ordering::SeqCst), 2);
+        })
+        .await;
+}

crates/scanner/src/scanner_io/tests/node_admission.rs

@@ -0,0 +1,136 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+use crate::workload_admission::ScannerNodeRequest;
+use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
+use std::num::NonZeroU16;
+use tokio::sync::oneshot;
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum ResumeCase {
+    Unchanged,
+    Updated,
+    Cancelled,
+}
+
+async fn check_cache_lease_resume(case: ResumeCase) {
+    let (_directory, store) = setup_local_scanner_store(1, 4).await;
+    let set = store.pools[0].disk_set[0].clone();
+    let source = DataUsageCacheSource::new(0, 0);
+    let cache_name = "node-resume/.usage-cache.bin";
+    let cache = DataUsageCache::default();
+    cache.save(set.clone(), cache_name).await.expect("seed cache");
+    let revisions = DataUsageCache::read_revisions(set.clone(), cache_name)
+        .await
+        .expect("initial revisions");
+    let gate = NodeAdmission::new(NonZeroU16::new(1).expect("one shared unit"));
+    gate.update(
+        &mut FeedbackController::new(),
+        Some(FeedbackObservation {
+            generation: 1,
+            max_concurrency: 1,
+            pressure: ResourcePressure::Unknown,
+            completed_units: None,
+        }),
+    );
+    let mut held = Some(gate.acquire().await);
+    let mut guard = acquire_scanner_cache_locks(set.as_ref(), cache_name, source)
+        .await
+        .expect("initial cache lease");
+    let ctx = CancellationToken::new();
+    let (respond, response) = oneshot::channel();
+    let request = ScannerNodeRequest {
+        owner: gate.clone(),
+        respond,
+        cancel: ctx.clone(),
+    };
+    let mut expected = revisions.clone();
+    let result = {
+        let resume = guard.resume_node_operation(set.clone(), cache_name, source, &revisions, request);
+        tokio::pin!(resume);
+        assert!(futures::poll!(&mut resume).is_pending(), "shared node is occupied");
+        let other = tokio::time::timeout(Duration::from_secs(1), acquire_scanner_cache_locks(set.as_ref(), cache_name, source))
+            .await
+            .expect("cache lock must be released before joining the node queue")
+            .expect("another cache worker can acquire the lease");
+        if case == ResumeCase::Updated {
+            let mut newer = cache;
+            newer.info.next_cycle = 1;
+            newer
+                .save(set.clone(), cache_name)
+                .await
+                .expect("another worker updates the cache");
+            expected = DataUsageCache::read_revisions(set.clone(), cache_name)
+                .await
+                .expect("new cache revisions");
+            assert_ne!(expected, revisions);
+        }
+        drop(other);
+        if case == ResumeCase::Cancelled {
+            ctx.cancel();
+        } else {
+            held.take().expect("held node unit").complete();
+        }
+        tokio::time::timeout(Duration::from_secs(2), resume)
+            .await
+            .expect("node resumption must finish or observe cancellation")
+    };
+    if case == ResumeCase::Unchanged {
+        result.expect("unchanged cache resumes the same walk");
+        let permit = response.await.expect("coordinator returns the node permit");
+        assert_eq!(gate.snapshot().active, 1);
+        assert_eq!(gate.completed(), Some(1), "admitting scanner work must not count it as completed");
+        assert!(!guard.is_lock_lost());
+        let lock = set
+            .new_ns_lock(RUSTFS_META_BUCKET, &scanner_cache_lock_resource(cache_name, source))
+            .await
+            .expect("competing cache lock");
+        assert!(
+            lock.get_write_lock_quiet(Duration::from_millis(20)).await.is_err(),
+            "cache lease must be restored before the operation resumes"
+        );
+        permit.complete();
+    } else {
+        result.expect_err("cancelled or stale work cannot resume");
+        assert!(response.await.is_err(), "failed resumption must not authorize storage work");
+        if case == ResumeCase::Cancelled {
+            assert!(guard.is_lock_lost());
+            assert_eq!(gate.snapshot().active, 1, "the running heal keeps its unit");
+            drop(held.take());
+            assert_eq!(gate.completed(), Some(0));
+        } else {
+            assert_eq!(gate.completed(), Some(1));
+        }
+    }
+    drop(guard);
+    assert_eq!(gate.snapshot().active, 0);
+    assert!(gate.try_acquire().is_some(), "dropped requests must leave no queue or capacity leak");
+    assert_eq!(
+        DataUsageCache::read_revisions(set.clone(), cache_name)
+            .await
+            .expect("final revisions"),
+        expected
+    );
+    acquire_scanner_cache_locks(set.as_ref(), cache_name, source)
+        .await
+        .expect("final cache lease is available");
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_wait_yields_cache_lease_before_fifo_admission() {
+    check_cache_lease_resume(ResumeCase::Unchanged).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_wait_rejects_cache_changed_while_unlocked() {
+    check_cache_lease_resume(ResumeCase::Updated).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_node_wait_cancellation_releases_queue_without_stealing_running_unit() {
+    check_cache_lease_resume(ResumeCase::Cancelled).await;
+}

crates/scanner/src/storage_api.rs

@@ -22,6 +22,8 @@ use serde::{Deserialize, Serialize};
 use tokio::sync::Notify;
 
 pub(crate) use rustfs_ecstore::with_background_disk_io as ecstore_with_background_disk_io;
+#[cfg(all(test, unix))]
+pub(crate) use rustfs_ecstore::with_background_resource_unit as ecstore_with_background_resource_unit;
 
 pub(crate) use s3s::dto::{
     BucketLifecycleConfiguration as EcstoreBucketLifecycleConfiguration, LifecycleRuleFilter as EcstoreLifecycleRuleFilter,
@@ -39,6 +41,8 @@ pub(crate) use rustfs_ecstore::api::data_usage::{NativeScannerPauseBacklogWriteF
 
 pub(crate) use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys as EcstoreBucketTargetSys;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc as EcstoreLcEventSrc;
+#[cfg(test)]
+pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::ExpiryState as EcstoreExpiryState;
 pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{
     apply_expiry_rule as ecstore_apply_expiry_rule, apply_transition_rule as ecstore_apply_transition_rule,
     lifecycle_version_delete_target as ecstore_lifecycle_version_delete_target,
@@ -99,6 +103,10 @@ pub(crate) use rustfs_ecstore::api::disk::{
 pub(crate) use rustfs_ecstore::api::disk::{
     DiskOption as EcstoreDiskOption, DiskStore as EcstoreDiskStore, new_disk as ecstore_new_disk,
 };
+#[cfg(all(test, unix))]
+pub(crate) use rustfs_ecstore::api::disk::{
+    LocalPublicationPause as EcstoreLocalPublicationPause, LocalPublicationStage as EcstoreLocalPublicationStage,
+};
 pub(crate) use rustfs_ecstore::api::error::{
     Error as EcstoreErrorType, Result as EcstoreResultType, StorageError as EcstoreStorageError,
 };
@@ -161,10 +169,17 @@ pub(crate) mod ecstore_diagnostic {
 pub(crate) type EcstoreHealResultItem = <EcstoreStore as storage_contracts::HealOperations>::HealResultItem;
 
 pub(crate) mod owner {
+    #[cfg(test)]
+    pub(crate) use super::EcstoreExpiryState;
     #[cfg(test)]
     pub(crate) use super::ecstore_set_object_store_resolver;
     #[cfg(test)]
     pub(crate) use super::storage_contracts::DiskSetSelector;
+    #[cfg(all(test, unix))]
+    pub(crate) use super::{
+        EcstoreLocalPublicationPause, EcstoreLocalPublicationStage, ecstore_with_background_disk_io,
+        ecstore_with_background_resource_unit,
+    };
     pub(crate) use super::{
         MAX_SCANNER_PAUSE_BACKLOG_BYTES, ScannerPauseBacklogRetirementError, ScannerPauseBacklogRetirementPlan,
         ScannerPauseBacklogRetirementReplica, register_scanner_pause_backlog_retirement_planner,

crates/scanner/src/workload_admission.rs

@@ -12,9 +12,143 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::{Arc, LazyLock, RwLock};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, LazyLock, OnceLock, RwLock};
 
+use rustfs_concurrency::adaptive::{NodeAdmission, NodePermit};
 use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshotProvider, WorkloadClass};
+use tokio::sync::{mpsc, oneshot};
+use tokio_util::sync::CancellationToken;
+
+use crate::ScannerError;
+
+static SCANNER_NODE_ADMISSION: OnceLock<Arc<NodeAdmission>> = OnceLock::new();
+
+tokio::task_local! {
+    static SCANNER_NODE_REQUESTS: ScannerNodeRequestSender;
+    static SCANNER_NODE_OPERATION_CANCEL: CancellationToken;
+}
+
+/// Cancellation for the currently admitted scanner operation, when present.
+pub(crate) fn scanner_node_operation_cancellation() -> Option<CancellationToken> {
+    SCANNER_NODE_OPERATION_CANCEL.try_with(Clone::clone).ok()
+}
+
+pub(crate) struct ScannerNodeRequest {
+    pub(crate) owner: Arc<NodeAdmission>,
+    pub(crate) respond: oneshot::Sender<NodePermit>,
+    pub(crate) cancel: CancellationToken,
+}
+
+pub(crate) type ScannerNodeRequestSender = mpsc::Sender<ScannerNodeRequest>;
+pub(crate) type ScannerNodeRequestReceiver = mpsc::Receiver<ScannerNodeRequest>;
+
+/// One suspended operation per walk; disabled admission allocates no channel.
+pub(crate) fn scanner_node_request_channel() -> (Option<ScannerNodeRequestSender>, Option<ScannerNodeRequestReceiver>) {
+    if scanner_node_admission().is_none() {
+        return (None, None);
+    }
+    let (sender, receiver) = mpsc::channel(1);
+    (Some(sender), Some(receiver))
+}
+
+pub(crate) async fn with_scanner_node_requests<F: std::future::Future>(
+    sender: Option<ScannerNodeRequestSender>,
+    operation: F,
+) -> F::Output {
+    match sender {
+        Some(sender) => SCANNER_NODE_REQUESTS.scope(sender, operation).await,
+        None => operation.await,
+    }
+}
+
+#[cfg(test)]
+tokio::task_local! {
+    pub(crate) static TEST_SCANNER_NODE_ADMISSION: Arc<NodeAdmission>;
+    pub(crate) static TEST_SCANNER_AFTER_NODE_ACQUIRE: Arc<dyn Fn() + Send + Sync>;
+}
+
+/// Install the heal manager's node owner before starting scanner workers.
+/// Refuse replacement: existing scans must not outlive an independently renewed owner.
+pub fn set_scanner_node_admission(owner: Arc<NodeAdmission>) -> bool {
+    SCANNER_NODE_ADMISSION.set(owner).is_ok()
+}
+
+pub(crate) fn scanner_node_admission() -> Option<Arc<NodeAdmission>> {
+    #[cfg(test)]
+    if let Ok(owner) = TEST_SCANNER_NODE_ADMISSION.try_with(Arc::clone) {
+        return Some(owner);
+    }
+    SCANNER_NODE_ADMISSION.get().cloned()
+}
+
+/// Try the fast path first. A scoped coordinator releases the cache lease before
+/// joining the node queue and validates the cache before returning a permit.
+/// Uncoordinated callers defer instead of parking under an unknown lock.
+pub(crate) async fn run_scanner_node_operation<T, E>(
+    owner: Option<&Arc<NodeAdmission>>,
+    cancel: &CancellationToken,
+    operation: impl std::future::Future<Output = Result<T, E>>,
+) -> Result<Result<T, E>, ScannerError> {
+    let Some(owner) = owner else {
+        return Ok(operation.await);
+    };
+    let operation = SCANNER_NODE_OPERATION_CANCEL.scope(cancel.clone(), operation);
+    if cancel.is_cancelled() {
+        return Err(ScannerError::Other("Operation cancelled".to_string()));
+    }
+    let permit = match owner.try_acquire() {
+        Some(permit) => permit,
+        None => {
+            let sender = SCANNER_NODE_REQUESTS
+                .try_with(Clone::clone)
+                .map_err(|_| ScannerError::ResourceDeferred)?;
+            let acquire = async {
+                let (respond, response) = oneshot::channel();
+                sender
+                    .send(ScannerNodeRequest {
+                        owner: owner.clone(),
+                        respond,
+                        cancel: cancel.clone(),
+                    })
+                    .await
+                    .map_err(|_| ScannerError::ResourceDeferred)?;
+                response.await.map_err(|_| ScannerError::ResourceDeferred)
+            };
+            tokio::select! {
+                biased;
+                _ = cancel.cancelled() => return Err(ScannerError::Other("Operation cancelled".to_string())),
+                permit = acquire => permit?,
+            }
+        }
+    };
+    #[cfg(test)]
+    let _ = TEST_SCANNER_AFTER_NODE_ACQUIRE.try_with(|hook| hook());
+    let operation_started = AtomicBool::new(false);
+    tokio::pin!(operation);
+    let observed = std::future::poll_fn(|context| {
+        operation_started.store(true, Ordering::Relaxed);
+        std::future::Future::poll(operation.as_mut(), context)
+    });
+    tokio::pin!(observed);
+    let result = tokio::select! {
+        biased;
+        _ = cancel.cancelled() => {
+            // The node slot belongs to the waiter, not the native operation.
+            drop(permit);
+            if !operation_started.load(Ordering::Relaxed) {
+                return Err(ScannerError::Other("Operation cancelled".to_string()));
+            }
+            // Keep polling the started operation under its existing locks.
+            return Ok(observed.await);
+        }
+        result = &mut observed => result,
+    };
+    if result.is_ok() && !cancel.is_cancelled() {
+        permit.complete();
+    }
+    Ok(result)
+}
 
 type WorkloadSnapshotProviderRef = Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
 

docs/architecture/README.md

@@ -50,6 +50,7 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 | [scanner-usage-authority-decision.md](scanner-usage-authority-decision.md) | deciding whether quota admission depends on scanner data usage, removing scanner publication layers, or designing a scanner storage boundary |
 | [scanner-per-bucket-convergence.md](scanner-per-bucket-convergence.md) | changing what makes a scanner cycle `Complete`, `Superseded`, or converged, adding or re-classifying a dirty-usage producer, deciding whether a background writeback may invalidate a cycle, changing scanner worker concurrency or throttling, or activating scoped dirty acknowledgements |
 | [config-model-boundary-adr.md](config-model-boundary-adr.md) | touching the server-config model (`Config`, `KV`, `KVS`) or its persistence, or asking which crate owns which part of server configuration |
+| [plugin-delegation.md](plugin-delegation.md) | changing original-user plugin delegation, attempt credentials, or Runner task APIs |
 | [admin-route-action-snapshot.md](admin-route-action-snapshot.md) | adding, moving, or re-authorizing an admin route and needing to know where the route → handler → `AdminAction` contract is enforced |
 | [kms-bulk-rekey-contract.md](kms-bulk-rekey-contract.md) | changing the bulk envelope re-wrap sweep, its admin endpoints, the re-wrap primitive, or which objects a rekey may touch |
 | [remote-credential-sealing-adr.md](remote-credential-sealing-adr.md) | adding, reading, or persisting a stored remote credential (replication target, remote tier, on-demand migration source), or needing the sealed-envelope format, the mixed-version rules, or the reason this is worth doing in one deployment and not in another |

docs/architecture/plugin-delegation.md

@@ -0,0 +1,142 @@
+# Plugin delegation
+
+**Use this when:** changing plugin delegation registration, revocation, broker identity, or the scope of attempt credentials.
+
+## Integration boundary
+
+Runner owns scheduling, leases, container execution and result publication. RustFS provides original-user delegation, scoped temporary credentials and authenticated task API forwarding. Inputs and outputs use ordinary versioned S3 buckets; event-driven jobs use the existing bucket notification/Webhook path. Managed artifact sealing, inventory, garbage collection and backup retention require separate storage contracts and are outside this integration.
+
+Set `RUSTFS_PLUGIN_RUNNER_URL` to the Controller HTTPS URL ending in `/v1/ingress/manual/jobs`, and `RUSTFS_PLUGIN_RUNNER_DEPLOYMENT` to the delegation deployment. Forwarding requires the published outbound mTLS client identity and uses the published trust roots when configured. Connections are reused within a TLS generation; publishing new TLS material replaces the pool, and removing the client identity rejects new forwarding requests.
+
+## Authority
+
+A delegation is an immutable restriction record owned by an ordinary IAM user. It is neither a permission grant nor a credential. Creating it requires that user's real signed request and `admin:CreatePluginDelegation`; reading and revoking require the corresponding `admin:GetPluginDelegation` and `admin:RevokePluginDelegation` actions. Root, temporary credentials and service accounts cannot own records.
+
+The record binds a deployment, an input bucket, literal input and output prefixes, and an expiry of at most 24 hours. The v1 profile writes to that same bucket; v2 explicitly names the output bucket. Wildcards, control characters and IAM variable expressions are rejected. Its random ID is also its immutable revision. There is no update operation.
+
+The Controller is responsible for validating accepted task identity, current lease, epoch, node session and control generation before requesting credentials. RustFS authenticates the configured broker and enforces the delegation and current parent IAM. RustFS does not independently verify a remote Controller lease.
+
+```mermaid
+sequenceDiagram
+    participant U as Original IAM user
+    participant C as Controller broker
+    participant R as RustFS
+    participant P as Plugin
+    U->>R: Signed delegation registration
+    R-->>U: Immutable ID and revision
+    C->>C: Validate task and current lease
+    C->>R: Signed issue-attempt request
+    R->>R: Broker mapping + active record + current IAM
+    R->>R: Persist scoped STS and recheck authority
+    R-->>C: Short-lived credentials
+    C-->>P: Attempt credentials
+    P->>R: Exact version GET / attempt-prefix PUT
+    R->>R: Current parent IAM intersected with session policy
+```
+
+## Persistence and revocation
+
+`crates/iam/src/delegation.rs` owns record validation and immutable revocation tombstones. Records use the existing IAM store under `config/iam/plugin-delegations/v1`. Writes require a successful readback. Revocation is idempotent and durable; it does not delete or overwrite the restriction record.
+
+`ObjectStore::load_iam_config_strict` distinguishes a missing object from an empty object, invalid JSON or failed decryption. Only a genuinely absent tombstone permits an active record. Corrupt or unavailable storage rejects access rather than treating revocation as absent. Existing non-delegation readers keep their compatibility behavior.
+
+Revocation prevents future issuance. It does not promise immediate invalidation of credentials already delivered; those remain subject to their expiration and current parent IAM. Registration, IAM changes and issuance are not one atomic transaction. Issuance rechecks after persistence and deletes the generated STS account if authority changed. Failed persistence or invalidation acknowledgment is reported as unknown issuance, never as safe retry permission.
+
+Manual job submission reads the active delegation before calling the Runner. A submission that passed this check is in flight and may be admitted if revocation races with the outbound request; revocation prevents submissions whose authority check occurs afterward, but does not cancel an admitted job. Callers must treat a successful submit receipt as accepted work even if the delegation is revoked later. After an ambiguous timeout or service error, retry the same idempotency key while the delegation remains active; after revocation or expiry, query status instead of retrying submission.
+
+`POST /rustfs/admin/v3/plugin-delegations/{id}/runs/query` accepts `{version:1,run_id}` and requires an ordinary IAM user with `admin:GetPluginRun` who owns the delegation in `{id}`. It reads the immutable owner record without checking expiry or revocation, then queries Runner using the record's delegation ID and revision and the authenticated access key. The result can contain only that caller's run status and committed manifest; it does not perform object reads or reauthorize object access. The Runner response must match the requested run ID and the strict status schema before RustFS returns it.
+
+`POST /rustfs/admin/v3/plugin-delegations/{id}/runs/cancel` accepts the same request and requires `admin:CancelPluginRun` plus ownership of `{id}`. Expiry and revocation do not block cancelling an existing run. Runner scopes the run to that owner; missing and other-owner runs share the same not-found response, while successful and failed runs return conflict. Success returns `{version:1,run_id,state:"CANCELLED",cleanup_pending}` and does not imply that asynchronous cleanup has finished.
+
+`POST /rustfs/admin/v3/plugin-delegations/{id}/runs/page` accepts `{version:1,limit,cursor?}` with a limit from 1 to 100 and requires `admin:ListPluginRuns` plus ownership of `{id}`. Expiry and revocation do not hide historical runs. RustFS supplies signer and delegation identity from authenticated IAM state; the Runner page is scoped to the exact caller and deployment. The strict response contains run summaries and an opaque-to-clients stable cursor; callers pass the returned cursor unchanged until it is null. RustFS bounds request/response sizes and rejects malformed, unordered, over-limit, or identity-invalid summaries.
+
+`POST /rustfs/admin/v3/plugin-delegations/{id}/runs/history` accepts `{version:1,run_id}` and requires `admin:ListPluginRuns` plus ownership of `{id}`. RustFS derives signer and delegation identity from the authenticated user and IAM record, then proxies Runner's bounded history response. The response contains at most the latest 32 committed state events in Runner's history-ID order and a `has_older` flag; timestamps are display data and are not used to reorder events. Missing and foreign runs share the same not-found response. RustFS rejects malformed, empty, over-limit, unexpected-state, or mismatched-run responses and bounds both request and response bodies.
+
+`POST /rustfs/admin/v3/plugin-delegations/{id}/runs/replay` accepts `{version:1,origin_run_id,idempotency_key,deadline_unix_micros}` and requires `admin:ReplayPluginRun`, current IAM authorization, ownership, and an active unexpired delegation. Caller identity fields are not accepted; RustFS derives signer and delegation revision from the authenticated request and IAM record. Runner accepts only an owned failed origin, preserves its fixed plugin/configuration/input version, and requires a new deadline no later than its configured run timeout. RustFS requires a positive Unix-microsecond deadline and verifies the strict receipt echoes it. Missing or foreign origins share not-found behavior; conflicts map to an operation conflict. Requests and receipts are bounded, and responses use `Cache-Control: no-store`.
+
+## Broker profile
+
+`rustfs/src/admin/handlers/plugin_attempt.rs` implements `POST /rustfs/admin/v3/plugin-delegations/{id}/issue-attempt`. The signer must be an ordinary IAM user with `admin:IssuePluginAttempt`, independently listed in `RUSTFS_PLUGIN_BROKERS`:
+
+```json
+{"broker-access-key":"deployment-name"}
+```
+
+This configuration contains public access keys, not secrets. It is immutable per process. Missing or invalid configuration disables broker issuance. The broker is a trusted Controller identity; ordinary users cannot substitute their own request fields for Controller validation.
+
+The request carries `tenant`, `run_id`, `attempt_id`, `revision`, `deployment`, `input_bucket`, `input_key`, `input_version`, `output_bucket`, `output_prefix` and `ttl_seconds`. IDs must be non-nil. Both profiles accept one exact input version and an output prefix equal to:
+
+```text
+<delegation-output-prefix><lowercase-hex-tenant>/<run-uuid>/<attempt-uuid>/
+```
+
+The session policy permits only `s3:GetObjectVersion` for that input with an exact `s3:VersionId`, and `s3:PutObject` beneath that attempt prefix. Credential lifetime is at most 900 seconds and cannot exceed delegation expiry. The Controller must additionally cap it to its validated lease and hard deadline.
+
+The original user's current IAM remains the parent permission boundary on every S3 request. The server's signing key only seals the token; it is not the plugin's authorization identity. Broker identity and original user identity are separate.
+
+External policy plugins must be disabled: their authorization path does not enforce this built-in session-policy boundary. Initializing, failed and ready plugin states all reject issuance. Attempt tokens carry a signed, persisted marker: after restart or on another node, marked credentials are accepted only by built-in STS authorization. A denied marked request returns a terminal error, so public bucket policies cannot bypass its session scope. The first profile also rejects parent policies containing conditions, because a broker request cannot reconstruct the original user's IP, MFA or other request context. It does not silently replace that context with the broker's.
+
+## Response and failures
+
+Successful issuance returns `version`, `delegation_id`, `delegation_revision`, `owner_access_key`, `access_key_id`, `secret_access_key`, `session_token` and `expires_at_unix_seconds`, with `Cache-Control: no-store`. The Controller verifies the returned delegation and owner against the accepted caller before delivery. Credentials must never enter logs, metrics or task result manifests.
+
+Missing records and records owned by a different user have the same read response. Revoked or expired records reject access. Storage failures remain errors. Post-persistence failures carry static `error_code`, `issued` and `issuance_unknown` facts without credentials; callers must reconcile an unknown outcome rather than automatically mint another credential.
+
+The route/action inventory is maintained in `rustfs/src/admin/route_policy.rs`. This contract does not imply event capture, physical container cleanup, restart reconciliation or upstream release acceptance.
+
+
+## Cross-bucket restriction profile v2
+
+The existing registration endpoint accepts this explicit v2 restriction:
+
+```json
+{
+  "version": 2,
+  "deployment": "runner-production",
+  "bucket": "source-images",
+  "input_prefix": "inputs/",
+  "output_bucket": "plugin-artifacts",
+  "output_prefix": "outputs/",
+  "expires_at_unix_seconds": 1893456000
+}
+```
+
+The example expiry must be replaced by a future value within the existing
+24-hour limit. `bucket` still identifies the input bucket. The existing
+`issue-attempt` request and credential response do not change: its
+`input_bucket` must equal `bucket`, and `output_bucket` must equal the explicit
+v2 output bucket. Neither a different bucket nor a sibling attempt prefix can
+be supplied by the broker. The exact input version and deterministic output
+prefix checks remain in force.
+
+A v1 restriction has neither `version` nor `output_bucket`; its JSON fields
+and same-bucket behavior are unchanged. V2 requires both `version: 2` and a
+nonempty, bounded, literal `output_bucket`. Missing pairs, explicit nulls,
+unknown versions and unknown fields are rejected rather than downgraded to v1.
+The immutable record envelope carries version 1 or 2 to match its restriction
+profile. Both remain in the existing IAM record namespace; no data migration
+or mutable record update is introduced. Readers reject mismatched envelope
+and restriction versions. Older readers reject v2, so deploy compatible
+readers before creating v2 records.
+
+V2 does not grant the original user additional permissions. Issuance retains
+its current-parent checks and post-persistence recheck/invalidation, and each
+S3 request still requires both the original user's current IAM policy and the
+attempt session policy to allow it. An explicit parent Deny or removal of
+output permission prevents writes even with an already issued token. No
+administrator credential substitutes for missing parent permission.
+
+Focused tests are `delegation_v1_json_remains_compatible`,
+`delegation_v2_requires_explicit_output_bucket`,
+`attempt_v2_policy_keeps_input_version_and_output_bucket` and
+`delegation_v2_persists_and_vends_with_real_parent_intersection`.
+The last uses the actual issuance handler, persistent IAM and four-disk
+storage, including a real input version and a third bucket allowed by the
+parent but forbidden by the delegation. Removing only the output-bucket
+comparison must fail `A03_CROSS_BUCKET_OUTPUT_BOUND`; parent intersection is
+checked independently with issued temporary credentials. These are internal
+handler/IAM tests, not signed HTTP or complete cross-bucket S3 execution.
+
+This profile adds no bucket-creation API, Runner binding support, credential
+endpoint or GC permission. Coordinating Runner's pinned binding with its
+result storage bucket remains a separate integration requirement.

docs/operations/scanner-runtime-controls.md

@@ -529,17 +529,21 @@ The OTLP exporter records these node-local histograms for completed manager admi
 
 The two new histograms aggregate all observed admission sources and results without per-request labels. An ordinary uninterrupted admission partitions the phase into wait and hold. Admin forceStart releases these guards while cancelling overlaps, then reacquires them: the cancellation gap is included only in the original total phase, and the second acquisition contributes to wait. Cancellation subroutine work is outside these outer admission guard intervals. Recording happens after all three guards are released. They measure manager admission ownership, not namespace/object-lock waiting during repair. The status field `maxLockPhaseMicros` remains a manager-lifetime maximum, not a p95 sample. Keep same-window histogram counts and bucket resolution when deriving percentiles; missing observations do not mean zero latency. These metrics do not establish performance acceptance by themselves.
 
-## Experimental heal node admission
+## Experimental shared node admission
 
-`RUSTFS_HEAL_ADAPTIVE_ADMISSION=true` enables Linux cgroup-v2 feedback for the four object-heal entry points owned by the heal manager. It is off by default while full SSD performance acceptance is pending. Unsupported platforms reject an enabled setting. This does not yet control scanner walks, bucket/format repair or physical devices, and it is not a replacement for existing worker, disk or byte budgets.
+`RUSTFS_HEAL_ADAPTIVE_ADMISSION=true` enables Linux cgroup-v2 feedback shared by the heal manager’s four object-heal entry points and local Scanner namespace reads and object accounting. It is off by default while full SSD performance acceptance is pending. Unsupported platforms reject an enabled setting. This does not yet control the abandoned-child streaming list coordinator, bucket/format repair or physical devices, and it is not a replacement for existing worker, disk or byte budgets.
 
 The node samples its effective CPU quota/cpuset, ancestor memory headroom, CPU usage/throttling and CPU/memory/I/O pressure once per second. Sampling uses one blocking worker at a time. Resource reads retain their real start/end times: reads spanning over 20 ms or older than one second after all inputs return cannot renew admission. Counter rates use both read brackets conservatively. The concurrency cap is at most the rounded-up effective CPU count and 256 scheduling units; existing operator limits still apply independently. At most 1% memory headroom blocks new work. This is a pressure safeguard, not a memory reservation or a claim that concurrent repair buffers fit in the remaining bytes.
 
 Each complete sample leases admission for three seconds. Missing resource caps block new work; missing or low-confidence feedback retains at most one maintenance unit. CPU use above 90%, throttling or pressure above 1% of elapsed time, memory headroom at most 10%, foreground errors/cancellations, a foreground queue or a comparable foreground p99 regression reduce concurrency immediately. Recovery requires CPU use at most 70%, throttling/stalls at most 0.1% of elapsed time, at least 20% memory headroom, fresh foreground evidence and actual successful work. The existing controller probes one additional unit only after eight consecutive healthy seconds and cooldown. Limits affect new admission and do not revoke active work.
 
+The Scanner captures the same node owner once per bucket walk. Directory open/refill, entry type and size lookup, and object accounting try admission before their first storage poll; each unit ends when its operation future finishes. On contention, local and remote coordinators suspend the same traversal, release the bucket cache lease, and join the shared FIFO queue. They reacquire the cache lease and verify both cache revisions before resuming. A changed cache rejects the suspended result. Cancellation also drops the queued request; the existing cycle budget still bounds the walk. This retains directory and recursion state while allowing Heal to advance at capacity one, without repeatedly replaying a partially classified directory.
+
+Calls without a coordinating cache lease retain immediate resource deferral and useful partial-cache handling. Existing remote single-session ownership and local-disk admission shares remain in force. Disabled admission creates no request channel. Enabled walks use one bounded channel, with a response allocation and cache revision reads only on contention; the uncontended per-entry path adds no global lookup or task. The streaming abandoned-child coordinator requires a separate ownership boundary: retaining a node unit while it waits for a bounded-channel consumer that needs the same unit would deadlock at capacity one.
+
 Foreground comparison uses the existing complete response-body windows and shared admission snapshots. Each active class needs at least 100 completions and a finite p99 interval; this permits the percentile rank but is not a statistical confidence guarantee. Comparisons require the same active classes and rates within 80–125% of a retained reference. A changed workload resets recovery. The reference is an observed workload window, not an independent background-off baseline. A p99 interval can justify stability only when its upper bound is within 5% of the reference lower bound; a lower bound above 105% of the reference upper bound is pressure. Overlapping/uncertain comparisons cannot justify growth. Missing queues, stale or repeated cuts, uncertain cut boundaries and low counts are not idle evidence.
 
-Admission waits at most 250 ms before returning the existing typed transient deferral, before polling storage or taking namespace locks. No-lock callers never wait. Queued read repair uses the same node gate; direct foreground ECStore calls remain outside this heal-manager adapter. Cancellation releases the non-cloneable node scheduling unit. Native work retains separate physical-I/O units, and unsuccessful/cancelled/dry-run work cannot increment the node's completed-work counter. This scheduling counter is not an authoritative heal receipt and cannot discharge repair responsibility.
+Heal admission waits at most 250 ms before returning the existing typed transient deferral, before polling storage or taking namespace locks. No-lock callers never wait. Queued read repair uses the same node gate; direct foreground ECStore calls remain outside this heal-manager adapter. Cancellation releases the non-cloneable node scheduling unit. Native work retains separate physical-I/O units, and unsuccessful/cancelled/dry-run work cannot increment the node's completed-work counter. This scheduling counter is not an authoritative heal receipt and cannot discharge repair responsibility.
 
 The `rustfs_background_node_admission_limit` and `rustfs_background_node_admission_active` gauges describe the last published sample, not a live reservation proof. A blocked sampler can leave these exports stale even though the admission lease has expired. Full disk-owner feedback, shared scanner scheduling, escaping-buffer byte ownership and the matched-release SSD matrix/hot-shrink/72-hour soak remain required before enabling this by default or claiming optimal performance.
 

rustfs/src/admin/handlers/mod.rs

@@ -52,6 +52,8 @@ pub mod object_data_cache;
 pub mod object_zip_download;
 pub mod oidc;
 pub mod on_demand_migration;
+pub mod plugin_attempt;
+pub mod plugin_delegation;
 pub mod plugins_catalog;
 pub mod plugins_instances;
 pub mod policies;

rustfs/src/admin/handlers/plugin_attempt.rs

@@ -0,0 +1,568 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+//! Restricted vending for explicitly configured brokers, never AssumeRole chaining.
+use super::plugin_delegation::{owner, response, storage_error};
+use crate::{
+    admin::{
+        router::{AdminOperation, Operation, S3Router},
+        runtime_sources,
+        storage_api::s3::{self, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result},
+    },
+    auth::check_key_valid_with_context,
+    server::ADMIN_PREFIX,
+};
+use http::{Method, StatusCode};
+use matchit::Params;
+use rustfs_iam::{
+    delegation::Delegation,
+    sys::{PLUGIN_ATTEMPT_CREDENTIAL_CLAIM, SESSION_POLICY_NAME},
+};
+use rustfs_policy::{
+    auth::get_new_credentials_with_metadata,
+    policy::{
+        Args, Policy,
+        action::{Action, AdminAction, S3Action},
+    },
+};
+use serde::{Deserialize, Serialize};
+use std::{
+    collections::HashMap,
+    sync::OnceLock,
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
+use time::OffsetDateTime;
+
+// Operator configuration contains public signer access keys, not credentials.
+// Empty/invalid configuration disables the endpoint. It is immutable per process.
+static BROKERS: OnceLock<Option<HashMap<String, String>>> = OnceLock::new();
+fn configured_deployment(key: &str) -> Option<&'static str> {
+    BROKERS
+        .get_or_init(|| {
+            let raw = std::env::var("RUSTFS_PLUGIN_BROKERS").ok()?;
+            let map: HashMap<String, String> = serde_json::from_str(&raw).ok()?;
+            if map.is_empty()
+                || map
+                    .iter()
+                    .any(|(k, v)| k.is_empty() || v.is_empty() || k.len() > 1024 || v.len() > 1024)
+            {
+                return None;
+            }
+            Some(map)
+        })
+        .as_ref()?
+        .get(key)
+        .map(String::as_str)
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Attempt {
+    tenant: String,
+    run_id: uuid::Uuid,
+    attempt_id: uuid::Uuid,
+    revision: String,
+    deployment: String,
+    input_bucket: String,
+    input_key: String,
+    input_version: String,
+    output_bucket: String,
+    output_prefix: String,
+    ttl_seconds: u64,
+}
+#[derive(Serialize)]
+struct Issued {
+    version: u32,
+    delegation_id: String,
+    delegation_revision: String,
+    owner_access_key: String,
+    access_key_id: String,
+    secret_access_key: String,
+    session_token: String,
+    expires_at_unix_seconds: i64,
+}
+
+pub fn register_plugin_attempt_route(router: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}/issue-attempt"),
+        AdminOperation(&Issue),
+    )?;
+    Ok(())
+}
+fn now() -> S3Result<u64> {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .map_err(|_| s3::error(S3ErrorCode::InternalError, "clock unavailable"))
+}
+fn literal(value: &str) -> bool {
+    !value.is_empty()
+        && value.len() <= 1024
+        && !value.contains("${")
+        && !value.chars().any(|c| c.is_control() || matches!(c, '*' | '?'))
+}
+fn session_policy(record: &Delegation, attempt: &Attempt) -> S3Result<Policy> {
+    let restriction = &record.restrictions;
+    // Stored records from earlier versions must obey the same literal rules.
+    restriction.validate(record.created_at_unix_seconds).map_err(storage_error)?;
+    let exact_output = format!(
+        "{}{}/{}/{}/",
+        restriction.output_prefix,
+        hex_simd::encode_to_string(attempt.tenant.as_bytes(), hex_simd::AsciiCase::Lower),
+        attempt.run_id,
+        attempt.attempt_id
+    );
+    if attempt.tenant.is_empty()
+        || attempt.tenant.len() > 1024
+        || attempt.run_id.is_nil()
+        || attempt.attempt_id.is_nil()
+        || attempt.output_prefix != exact_output
+        || attempt.revision != record.revision
+        || attempt.deployment != restriction.deployment
+        || attempt.input_bucket != restriction.bucket
+        || attempt.output_bucket != restriction.output_bucket().map_err(storage_error)?
+        || !literal(&attempt.input_key)
+        || !attempt.input_key.starts_with(&restriction.input_prefix)
+        || !literal(&attempt.input_version)
+        || attempt.input_version == "null"
+        || !literal(&attempt.output_prefix)
+        || !attempt.output_prefix.starts_with(&restriction.output_prefix)
+        || attempt.output_prefix == restriction.output_prefix
+        || !attempt.output_prefix.ends_with('/')
+        || !(1..=900).contains(&attempt.ttl_seconds)
+    {
+        return Err(s3::error(S3ErrorCode::AccessDenied, "attempt restriction rejected"));
+    }
+    let value = serde_json::json!({"Version":"2012-10-17","Statement":[
+        {"Effect":"Allow","Action":["s3:GetObjectVersion"],"Resource":[format!("arn:aws:s3:::{}/{}",attempt.input_bucket,attempt.input_key)],"Condition":{"StringEquals":{"s3:VersionId":attempt.input_version}}},
+        {"Effect":"Allow","Action":["s3:PutObject"],"Resource":[format!("arn:aws:s3:::{}/{}*",attempt.output_bucket,attempt.output_prefix)]}
+    ]});
+    // Condition operators borrow their keys while the policy is decoded.
+    let bytes = serde_json::to_vec(&value).map_err(|_| s3::error(S3ErrorCode::InternalError, "attempt policy unavailable"))?;
+    serde_json::from_slice(&bytes).map_err(|_| s3::error(S3ErrorCode::InternalError, "attempt policy unavailable"))
+}
+
+fn issuance_error(code: &'static str, issued: bool, unknown: bool) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let mut answer = response(&serde_json::json!({"error_code":code,"issued":issued,"issuance_unknown":unknown}))?;
+    answer.output.0 = StatusCode::SERVICE_UNAVAILABLE;
+    Ok(answer)
+}
+fn same_parent_authority(before_names: &[String], before_policy: &Policy, after_names: &[String], after_policy: &Policy) -> bool {
+    fn names(names: &[String]) -> Vec<&str> {
+        let mut names: Vec<_> = names.iter().map(String::as_str).collect();
+        names.sort_unstable();
+        names.dedup();
+        names
+    }
+    fn statements(policy: &Policy) -> Option<Vec<Vec<u8>>> {
+        let mut statements = policy
+            .statements
+            .iter()
+            .map(serde_json::to_vec)
+            .collect::<Result<Vec<_>, _>>()
+            .ok()?;
+        statements.sort_unstable();
+        Some(statements)
+    }
+
+    names(before_names) == names(after_names)
+        && matches!(
+            (statements(before_policy), statements(after_policy)),
+            (Some(before), Some(after)) if before == after
+        )
+}
+async fn require_builtin_authorization() -> S3Result<()> {
+    rustfs_iam::sys::IamSys::<rustfs_iam::store::object::ObjectStore>::ensure_restricted_credential_vending()
+        .await
+        .map_err(|_| s3::error(S3ErrorCode::AccessDenied, "external policy plugin unsupported for attempt credentials"))
+}
+
+struct Issue;
+#[async_trait::async_trait]
+impl Operation for Issue {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        require_builtin_authorization().await?;
+        let (iam, broker) = owner(&req, AdminAction::IssuePluginAttemptAction).await?;
+        let deployment =
+            configured_deployment(&broker).ok_or_else(|| s3::error(S3ErrorCode::AccessDenied, "broker deployment rejected"))?;
+        let context = runtime_sources::app_context_from_req(&req)
+            .ok_or_else(|| s3::error(S3ErrorCode::InternalError, "application context unavailable"))?;
+        let bytes = tokio::time::timeout(Duration::from_secs(10), req.input.store_all_limited(16384))
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "attempt body unavailable"))?
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "attempt body unavailable"))?;
+        let attempt: Attempt =
+            serde_json::from_slice(&bytes).map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid attempt request"))?;
+        if attempt.deployment != deployment {
+            return Err(s3::error(S3ErrorCode::AccessDenied, "broker deployment rejected"));
+        }
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        let record = iam
+            .read_active_plugin_delegation_for_broker(id, now()?)
+            .await
+            .map_err(storage_error)?;
+        let policy = session_policy(&record, &attempt)?;
+        let (principal, root) = check_key_valid_with_context("", &record.owner_access_key, Some(context.as_ref())).await?;
+        if root || principal.is_temp() || principal.is_service_account() {
+            return Err(s3::error(S3ErrorCode::AccessDenied, "original user required"));
+        }
+        // A broker request cannot reconstruct the original user's IP/MFA/tag
+        // conditions. This first profile rejects conditional parent policies.
+        let policies = iam
+            .policy_db_get(&principal.access_key, &principal.groups)
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InternalError, "parent policy unavailable"))?;
+        let parent_policy = iam.get_combined_policy(&policies).await;
+        if policies.is_empty() || parent_policy.statements.iter().any(|s| !s.conditions.is_empty()) {
+            return Err(s3::error(S3ErrorCode::AccessDenied, "parent condition context unsupported"));
+        }
+        let mut conditions = HashMap::new();
+        conditions.insert("s3:VersionId".to_owned(), vec![attempt.input_version.clone()]);
+        if !iam
+            .is_allowed(&Args {
+                account: &principal.access_key,
+                groups: &principal.groups,
+                action: Action::S3Action(S3Action::GetObjectVersionAction),
+                bucket: &attempt.input_bucket,
+                object: &attempt.input_key,
+                conditions: &conditions,
+                is_owner: false,
+                claims: principal.claims_or_empty(),
+                deny_only: false,
+            })
+            .await
+        {
+            return Err(s3::error(S3ErrorCode::AccessDenied, "parent input rejected"));
+        }
+        let expiration = now()?
+            .checked_add(attempt.ttl_seconds)
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "invalid expiration"))?
+            .min(record.restrictions.expires_at_unix_seconds);
+        let expires_at = OffsetDateTime::from_unix_timestamp(
+            i64::try_from(expiration).map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid expiration"))?,
+        )
+        .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid expiration"))?;
+        let mut claims = principal.claims.clone().unwrap_or_default();
+        claims.insert(PLUGIN_ATTEMPT_CREDENTIAL_CLAIM.into(), serde_json::Value::Bool(true));
+        claims.insert("exp".into(), serde_json::json!(expires_at.unix_timestamp()));
+        claims.insert("parent".into(), serde_json::json!(principal.access_key));
+        let bytes =
+            serde_json::to_vec(&policy).map_err(|_| s3::error(S3ErrorCode::InternalError, "attempt policy unavailable"))?;
+        claims.insert(
+            SESSION_POLICY_NAME.into(),
+            serde_json::json!(base64_simd::URL_SAFE_NO_PAD.encode_to_string(&bytes)),
+        );
+        let fresh = iam
+            .read_active_plugin_delegation_for_broker(id, now()?)
+            .await
+            .map_err(storage_error)?;
+        if fresh != record {
+            return Err(s3::error(S3ErrorCode::AccessDenied, "attempt authority changed"));
+        }
+        owner(&req, AdminAction::IssuePluginAttemptAction).await?;
+        require_builtin_authorization().await?;
+        let signing = context
+            .action_credentials()
+            .get()
+            .ok_or_else(|| s3::error(S3ErrorCode::InternalError, "token signer unavailable"))?;
+        // The server key seals the token only. Authorization above uses the
+        // original user's current IAM, and S3 continues its parent intersection.
+        let mut credential = get_new_credentials_with_metadata(&claims, &signing.secret_key)
+            .map_err(|_| s3::error(S3ErrorCode::InternalError, "attempt credential unavailable"))?;
+        credential.parent_user = principal.access_key.clone();
+        if iam.set_temp_user(&credential.access_key, &credential, None).await.is_err() {
+            return issuance_error("PLUGIN_ATTEMPT_ISSUANCE_UNKNOWN", false, true);
+        }
+        // No atomic IAM/delegation transaction spans persistence. Recheck and
+        // invalidate the generated credential if any authority changed.
+        let valid = async {
+            require_builtin_authorization().await?;
+            let after = iam
+                .read_active_plugin_delegation_for_broker(id, now()?)
+                .await
+                .map_err(storage_error)?;
+            let (current, current_root) =
+                check_key_valid_with_context("", &record.owner_access_key, Some(context.as_ref())).await?;
+            let names = iam
+                .policy_db_get(&current.access_key, &current.groups)
+                .await
+                .map_err(|_| s3::error(S3ErrorCode::InternalError, "parent policy unavailable"))?;
+            let current_policy = iam.get_combined_policy(&names).await;
+            if after != record
+                || current_root
+                || current.is_temp()
+                || current.is_service_account()
+                || current.secret_key != principal.secret_key
+                || !same_parent_authority(&policies, &parent_policy, &names, &current_policy)
+            {
+                return Err(s3::error(S3ErrorCode::AccessDenied, "attempt authority changed"));
+            }
+            // Re-evaluate the broker against its real signed request context.
+            owner(&req, AdminAction::IssuePluginAttemptAction).await?;
+            Ok::<(), S3Error>(())
+        }
+        .await;
+        if valid.is_err() {
+            if iam.delete_temp_account(&credential.access_key, true).await.is_err() {
+                return issuance_error("PLUGIN_ATTEMPT_INVALIDATION_UNKNOWN", true, true);
+            }
+            return issuance_error("PLUGIN_ATTEMPT_POST_ISSUANCE_REJECTED", true, false);
+        }
+        response(&Issued {
+            version: 1,
+            delegation_id: record.id,
+            delegation_revision: record.revision,
+            owner_access_key: record.owner_access_key,
+            access_key_id: credential.access_key,
+            secret_access_key: credential.secret_key,
+            session_token: credential.session_token,
+            expires_at_unix_seconds: expires_at.unix_timestamp(),
+        })
+    }
+}
+
+#[cfg(test)]
+#[path = "plugin_attempt_v2_tests.rs"]
+mod v2_tests;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn parent_authority_recheck_ignores_policy_order_but_detects_changes() {
+        let first: Policy = serde_json::from_value(serde_json::json!({
+            "Version":"2012-10-17",
+            "Statement":[
+                {"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/input/*"},
+                {"Effect":"Allow","Action":"s3:PutObject","Resource":"arn:aws:s3:::photos/output/*"}
+            ]
+        }))
+        .unwrap();
+        let reordered: Policy = serde_json::from_value(serde_json::json!({
+            "Version":"2012-10-17",
+            "Statement":[
+                {"Effect":"Allow","Action":"s3:PutObject","Resource":"arn:aws:s3:::photos/output/*"},
+                {"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/input/*"}
+            ]
+        }))
+        .unwrap();
+        let before_names = vec!["owner-policy".to_owned(), "bucket-policy".to_owned()];
+        let after_names = vec!["bucket-policy".to_owned(), "owner-policy".to_owned()];
+        assert!(same_parent_authority(&before_names, &first, &after_names, &reordered));
+
+        let changed: Policy = serde_json::from_value(serde_json::json!({
+            "Version":"2012-10-17",
+            "Statement":[
+                {"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/input/*"},
+                {"Effect":"Allow","Action":"s3:DeleteObject","Resource":"arn:aws:s3:::photos/output/*"}
+            ]
+        }))
+        .unwrap();
+        assert!(!same_parent_authority(&before_names, &first, &after_names, &changed));
+        let changed_resource: Policy = serde_json::from_value(serde_json::json!({
+            "Version":"2012-10-17",
+            "Statement":[
+                {"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/input/*"},
+                {"Effect":"Allow","Action":"s3:PutObject","Resource":"arn:aws:s3:::photos/private/*"}
+            ]
+        }))
+        .unwrap();
+        assert!(!same_parent_authority(&before_names, &first, &after_names, &changed_resource));
+        let added_deny: Policy = serde_json::from_value(serde_json::json!({
+            "Version":"2012-10-17",
+            "Statement":[
+                {"Effect":"Deny","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/input/private/*"},
+                {"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::photos/input/*"},
+                {"Effect":"Allow","Action":"s3:PutObject","Resource":"arn:aws:s3:::photos/output/*"}
+            ]
+        }))
+        .unwrap();
+        assert!(!same_parent_authority(&before_names, &first, &after_names, &added_deny));
+        assert!(!same_parent_authority(&before_names, &first, &["other-policy".to_owned()], &first));
+    }
+    #[tokio::test]
+    async fn attempt_policy_keeps_exact_version_and_narrow_output() {
+        let record = Delegation {
+            version: 1,
+            id: "a".repeat(32),
+            revision: "a".repeat(32),
+            owner_access_key: "user".into(),
+            created_at_unix_seconds: 1,
+            restrictions: rustfs_iam::delegation::Restrictions {
+                version: None,
+                output_bucket: None,
+                deployment: "runner".into(),
+                bucket: "photos".into(),
+                input_prefix: "input/".into(),
+                output_prefix: "output/".into(),
+                expires_at_unix_seconds: 1000,
+            },
+        };
+        let run_id = uuid::Uuid::new_v4();
+        let attempt_id = uuid::Uuid::new_v4();
+        let output_prefix = format!(
+            "output/{}/{}/{}/",
+            hex_simd::encode_to_string(b"tenant", hex_simd::AsciiCase::Lower),
+            run_id,
+            attempt_id
+        );
+        let mut attempt = Attempt {
+            tenant: "tenant".into(),
+            run_id,
+            attempt_id,
+            revision: record.revision.clone(),
+            deployment: "runner".into(),
+            input_bucket: "photos".into(),
+            input_key: "input/photo.png".into(),
+            input_version: "version-1".into(),
+            output_bucket: "photos".into(),
+            output_prefix: output_prefix.clone(),
+            ttl_seconds: 60,
+        };
+        let policy = session_policy(&record, &attempt).unwrap();
+        let scoped_output = format!("{output_prefix}result");
+        let groups = None;
+        let claims = HashMap::new();
+        for (action, object, version, expected) in [
+            (S3Action::GetObjectVersionAction, "input/photo.png", Some("version-1"), true),
+            (S3Action::GetObjectVersionAction, "input/photo.png", Some("version-2"), false),
+            (S3Action::GetObjectVersionAction, "input/photo.png", None, false),
+            (S3Action::GetObjectAction, "input/photo.png", None, false),
+            (S3Action::PutObjectAction, scoped_output.as_str(), None, true),
+            (S3Action::PutObjectAction, "outside/result", None, false),
+            (S3Action::PutObjectAction, "output/result", None, false),
+        ] {
+            let conditions = version
+                .map(|value| HashMap::from([("versionid".into(), vec![value.into()])]))
+                .unwrap_or_default();
+            assert_eq!(
+                policy
+                    .is_allowed(&Args {
+                        account: "user",
+                        groups: &groups,
+                        action: Action::S3Action(action),
+                        bucket: "photos",
+                        object,
+                        conditions: &conditions,
+                        is_owner: false,
+                        claims: &claims,
+                        deny_only: false,
+                    })
+                    .await,
+                expected,
+                "attempt policy action={action:?} object={object} version={version:?}"
+            );
+        }
+        attempt.output_bucket = "artifacts".into();
+        assert!(
+            session_policy(&record, &attempt).is_err(),
+            "v1 still requires the input bucket for output"
+        );
+        attempt.output_bucket = "photos".into();
+        attempt.output_prefix = "output/".into();
+        assert!(session_policy(&record, &attempt).is_err());
+        attempt.output_prefix = output_prefix;
+        attempt.input_key = "outside/photo.png".into();
+        assert!(session_policy(&record, &attempt).is_err());
+    }
+    #[tokio::test]
+    async fn attempt_v2_policy_keeps_input_version_and_output_bucket() {
+        let record: Delegation = serde_json::from_value(serde_json::json!({
+            "version":2,"id":"a".repeat(32),"revision":"a".repeat(32),"owner_access_key":"user",
+            "created_at_unix_seconds":1,"restrictions":{"version":2,"deployment":"runner","bucket":"photos",
+            "input_prefix":"input/","output_bucket":"artifacts","output_prefix":"output/","expires_at_unix_seconds":1000}
+        }))
+        .expect("explicit v2 delegation");
+        let run_id = uuid::Uuid::new_v4();
+        let attempt_id = uuid::Uuid::new_v4();
+        let output = format!("output/74656e616e74/{run_id}/{attempt_id}/");
+        let mut attempt = Attempt {
+            tenant: "tenant".into(),
+            run_id,
+            attempt_id,
+            revision: record.revision.clone(),
+            deployment: "runner".into(),
+            input_bucket: "photos".into(),
+            input_key: "input/photo.png".into(),
+            input_version: "version-1".into(),
+            output_bucket: "artifacts".into(),
+            output_prefix: output.clone(),
+            ttl_seconds: 60,
+        };
+        let policy = session_policy(&record, &attempt).expect("v2 scope");
+        let result = format!("{output}result");
+        for (action, bucket, object, version, expected) in [
+            (S3Action::GetObjectVersionAction, "photos", "input/photo.png", Some("version-1"), true),
+            (S3Action::GetObjectVersionAction, "photos", "input/photo.png", Some("version-2"), false),
+            (S3Action::GetObjectVersionAction, "artifacts", "input/photo.png", Some("version-1"), false),
+            (S3Action::PutObjectAction, "artifacts", result.as_str(), None, true),
+            (S3Action::PutObjectAction, "photos", result.as_str(), None, false),
+            (S3Action::PutObjectAction, "third", result.as_str(), None, false),
+            (S3Action::PutObjectAction, "artifacts", "output/other/result", None, false),
+        ] {
+            let conditions = version
+                .map(|v| HashMap::from([("versionid".into(), vec![v.into()])]))
+                .unwrap_or_default();
+            assert_eq!(
+                policy
+                    .is_allowed(&Args {
+                        account: "user",
+                        groups: &None,
+                        action: Action::S3Action(action),
+                        bucket,
+                        object,
+                        conditions: &conditions,
+                        claims: &HashMap::new(),
+                        is_owner: false,
+                        deny_only: false
+                    })
+                    .await,
+                expected,
+                "A03_CROSS_BUCKET_SCOPE: bucket={bucket} object={object}"
+            );
+        }
+        attempt.output_bucket = "third".into();
+        assert!(session_policy(&record, &attempt).is_err(), "A03_CROSS_BUCKET_OUTPUT_BOUND");
+        attempt.output_bucket = "photos".into();
+        assert!(session_policy(&record, &attempt).is_err(), "A03_CROSS_BUCKET_OUTPUT_BOUND");
+        attempt.output_bucket = "artifacts".into();
+        attempt.input_bucket = "artifacts".into();
+        assert!(session_policy(&record, &attempt).is_err(), "A03_CROSS_BUCKET_INPUT_BOUND");
+    }
+
+    #[tokio::test]
+    async fn actual_policy_variable_resolution_is_rejected_by_literal_scope() {
+        let resource = rustfs_policy::policy::resource::Resource::S3("photos/input/${aws:username}.png".into());
+        let conditions = HashMap::from([("username".into(), vec!["alice".into()])]);
+        let resolver =
+            rustfs_policy::policy::variables::VariableResolver::new(rustfs_policy::policy::variables::VariableContext {
+                username: Some("alice".into()),
+                ..Default::default()
+            });
+        assert!(
+            resource
+                .is_match_with_resolver("photos/input/alice.png", &conditions, Some(&resolver))
+                .await,
+            "C07_LITERAL_SCOPE_BOUND: actual policy engine substitutes resource variables"
+        );
+        assert!(
+            !literal("input/${aws:username}.png"),
+            "C07_LITERAL_SCOPE_BOUND: requested resources must stay literal"
+        );
+        let restriction = rustfs_iam::delegation::Restrictions {
+            version: None,
+            output_bucket: None,
+            deployment: "runner".into(),
+            bucket: "photos".into(),
+            input_prefix: "input/${aws:username}/".into(),
+            output_prefix: "output/".into(),
+            expires_at_unix_seconds: 200,
+        };
+        assert!(
+            restriction.validate(100).is_err(),
+            "C07_LITERAL_SCOPE_BOUND: legacy stored variable restrictions are rejected"
+        );
+    }
+}

rustfs/src/admin/handlers/plugin_attempt_v2_tests.rs

@@ -0,0 +1,365 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+//! Real handler, persisted IAM and four-disk storage; this is not a signed HTTP/S3 end-to-end test.
+use super::*;
+use crate::admin::{
+    runtime_sources::{AppContext, IamInterface, KmsInterface, ServerContextSlot},
+    storage_api::{
+        ECStore,
+        contract::bucket::{BucketOperations as _, MakeBucketOptions},
+        contract::object::ObjectIO as _,
+    },
+};
+use crate::storage::storage_api::{StorageObjectOptions, StoragePutObjReader};
+use http_body_util::BodyExt as _;
+use rustfs_iam::{
+    store::{
+        Store as _, UserType,
+        object::{IAM_CONFIG_PREFIX, ObjectStore},
+    },
+    sys::IamSys,
+};
+use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
+use serde_json::{Value, json};
+use std::sync::Arc;
+
+const PARENT: &str = "cross-parent";
+const BROKER: &str = "cross-broker";
+const SECRET: &str = "cross-bucket-test-secret";
+const ROOT: &str = "cross-root";
+const ROOT_SECRET: &str = "cross-bucket-root-test-secret";
+
+struct RequestIam(Arc<IamSys<ObjectStore>>);
+impl IamInterface for RequestIam {
+    fn handle(&self) -> Arc<IamSys<ObjectStore>> {
+        self.0.clone()
+    }
+    fn is_ready(&self) -> bool {
+        true
+    }
+}
+struct RequestKms;
+impl KmsInterface for RequestKms {
+    fn handle(&self) -> Arc<rustfs_kms::KmsServiceManager> {
+        Arc::new(rustfs_kms::KmsServiceManager::new())
+    }
+}
+struct Fixture {
+    _root: tempfile::TempDir,
+    store: Arc<ECStore>,
+    iam: Arc<IamSys<ObjectStore>>,
+    context: Arc<AppContext>,
+    input_version: String,
+}
+impl Fixture {
+    async fn new() -> Self {
+        let root = tempfile::tempdir().expect("cross-bucket storage root");
+        let env = rustfs_test_utils::TestECStoreEnv::builder()
+            .base_dir(root.path())
+            .disk_count(4)
+            .build()
+            .await;
+        assert_eq!(env.disk_paths.len(), 4);
+        let store = env.ecstore;
+        for bucket in ["photos", "artifacts", "third"] {
+            store
+                .make_bucket(
+                    bucket,
+                    &MakeBucketOptions {
+                        versioning_enabled: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("real versioned bucket");
+        }
+        let object = store
+            .put_object(
+                "photos",
+                "input/photo.png",
+                &mut StoragePutObjReader::from_vec(vec![0xA5; 273]),
+                &StorageObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("real input version");
+        let input_version = object.version_id.expect("exact input version").to_string();
+        ObjectStore::new(store.clone())
+            .save_iam_config(json!({"version":1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
+            .await
+            .expect("persistent IAM format");
+        let iam = rustfs_iam::build_iam_sys(store.clone()).await.expect("real IAM");
+        let context = Arc::new(AppContext::new(store.clone(), Arc::new(RequestIam(iam.clone())), Arc::new(RequestKms)));
+        assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
+            access_key: ROOT.into(),
+            secret_key: ROOT_SECRET.into(),
+            status: "on".into(),
+            ..Default::default()
+        }));
+        let fixture = Self {
+            _root: root,
+            store,
+            iam,
+            context,
+            input_version,
+        };
+        fixture.user(PARENT, Self::parent_policy(true, false)).await;
+        fixture
+            .user(
+                BROKER,
+                json!({"Version":"2012-10-17","Statement":[{
+            "Effect":"Allow","Action":"admin:IssuePluginAttempt"}]}),
+            )
+            .await;
+        fixture
+    }
+    fn parent_policy(output: bool, deny: bool) -> Value {
+        let mut resources = vec!["arn:aws:s3:::photos/output/*", "arn:aws:s3:::third/output/*"];
+        if output {
+            resources.push("arn:aws:s3:::artifacts/output/*");
+        }
+        let mut statements = vec![
+            json!({"Effect":"Allow","Action":"s3:GetObjectVersion","Resource":"arn:aws:s3:::photos/input/*"}),
+            json!({"Effect":"Allow","Action":"s3:PutObject","Resource":resources}),
+        ];
+        if deny {
+            statements.push(json!({"Effect":"Deny","Action":"s3:PutObject","Resource":"arn:aws:s3:::artifacts/*"}));
+        }
+        json!({"Version":"2012-10-17","Statement":statements})
+    }
+    async fn policy(&self, name: &str, value: Value) {
+        let policy = Policy::parse_config(&serde_json::to_vec(&value).expect("policy JSON")).expect("real IAM policy");
+        self.iam.set_policy(name, policy).await.expect("persist policy");
+    }
+    async fn user(&self, name: &str, policy: Value) {
+        self.iam
+            .create_user(
+                name,
+                &AddOrUpdateUserReq {
+                    secret_key: SECRET.into(),
+                    policy: None,
+                    status: AccountStatus::Enabled,
+                },
+            )
+            .await
+            .expect("create real user");
+        self.policy(name, policy).await;
+        self.iam
+            .policy_db_set(name, UserType::Reg, false, name)
+            .await
+            .expect("attach actual IAM policy");
+    }
+    fn request(&self, body: &Value) -> S3Request<Body> {
+        let slot = ServerContextSlot::new();
+        assert!(slot.install(self.context.clone()));
+        let mut req = S3Request {
+            input: Body::from(serde_json::to_vec(body).expect("attempt JSON")),
+            method: Method::POST,
+            uri: http::Uri::default(),
+            headers: http::HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+        req.extensions.insert(slot);
+        req.method = Method::POST;
+        req.credentials = Some(s3::auth::Credentials {
+            access_key: BROKER.into(),
+            secret_key: s3::auth::SecretKey::from(SECRET.to_owned()),
+        });
+        req
+    }
+    async fn issue(&self, record: &Delegation, body: &Value) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let mut router = matchit::Router::new();
+        router.insert("/{id}", ()).expect("parameter route");
+        let path = format!("/{}", record.id);
+        Issue
+            .call(self.request(body), router.at(&path).expect("actual delegation ID").params)
+            .await
+    }
+    async fn allowed(
+        &self,
+        credential: &rustfs_credentials::Credentials,
+        action: S3Action,
+        bucket: &str,
+        object: &str,
+        version: Option<&str>,
+    ) -> bool {
+        let conditions = version
+            .map(|v| HashMap::from([("versionid".into(), vec![v.into()])]))
+            .unwrap_or_default();
+        self.iam
+            .is_allowed(&Args {
+                account: &credential.access_key,
+                groups: &credential.groups,
+                action: Action::S3Action(action),
+                bucket,
+                object,
+                conditions: &conditions,
+                claims: credential.claims_or_empty(),
+                is_owner: false,
+                deny_only: false,
+            })
+            .await
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[serial_test::serial]
+async fn delegation_v2_persists_and_vends_with_real_parent_intersection() {
+    temp_env::async_with_vars(
+        [
+            ("RUSTFS_PLUGIN_BROKERS", Some(r#"{"cross-broker":"runner"}"#)),
+            ("RUSTFS_SKIP_BACKGROUND_TASK", Some("1")),
+        ],
+        async {
+            let fixture = Fixture::new().await;
+            let now = now().expect("clock");
+            let legacy = json!({"deployment":"runner","bucket":"photos","input_prefix":"input/",
+            "output_prefix":"output/","expires_at_unix_seconds":now+3600});
+            let v1 = fixture
+                .iam
+                .create_plugin_delegation(PARENT, serde_json::from_value(legacy.clone()).expect("v1"), now)
+                .await
+                .expect("persist legacy delegation");
+            assert_eq!(v1.version, 1);
+            assert_eq!(serde_json::to_value(&v1.restrictions).expect("v1 wire"), legacy);
+            let mut v2 = legacy;
+            v2["version"] = json!(2);
+            v2["output_bucket"] = json!("artifacts");
+            let record = fixture
+                .iam
+                .create_plugin_delegation(PARENT, serde_json::from_value(v2.clone()).expect("v2"), now)
+                .await
+                .expect("persist v2 delegation");
+            assert_eq!(record.version, 2);
+            let loaded = fixture
+                .iam
+                .read_active_plugin_delegation(PARENT, &record.id, now)
+                .await
+                .expect("strict stored v2");
+            assert!(loaded == record, "same persisted v2 identity");
+            assert_eq!(serde_json::to_value(&loaded.restrictions).expect("v2 wire"), v2);
+            // The record envelope and restriction profile cannot disagree on a persisted read.
+            let path = format!("config/iam/plugin-delegations/v1/{}/record.json", record.id);
+            let mut corrupt = serde_json::to_value(&record).expect("record JSON");
+            corrupt["version"] = json!(1);
+            ObjectStore::new(fixture.store.clone())
+                .save_iam_config(corrupt, &path)
+                .await
+                .expect("inject incompatible stored version");
+            assert!(
+                fixture
+                    .iam
+                    .read_active_plugin_delegation(PARENT, &record.id, now)
+                    .await
+                    .is_err(),
+                "mixed v1/v2 stored profiles fail closed"
+            );
+            ObjectStore::new(fixture.store.clone())
+                .save_iam_config(&record, &path)
+                .await
+                .expect("restore exact stored record");
+            let run = uuid::Uuid::new_v4();
+            let attempt = uuid::Uuid::new_v4();
+            let prefix = format!("output/74656e616e74/{run}/{attempt}/");
+            let body = json!({"tenant":"tenant","run_id":run,"attempt_id":attempt,"revision":record.revision,
+            "deployment":"runner","input_bucket":"photos","input_key":"input/photo.png","input_version":fixture.input_version,
+            "output_bucket":"artifacts","output_prefix":prefix,"ttl_seconds":60});
+            let loaded_v1 = fixture
+                .iam
+                .read_active_plugin_delegation(PARENT, &v1.id, now)
+                .await
+                .expect("read legacy record");
+            assert!(loaded_v1 == v1, "legacy persistent identity");
+            let mut legacy_body = body.clone();
+            legacy_body["revision"] = json!(v1.revision);
+            legacy_body["output_bucket"] = json!("photos");
+            let legacy_response = fixture.issue(&v1, &legacy_body).await.expect("legacy same-bucket issuance");
+            assert_eq!(legacy_response.output.0, StatusCode::OK);
+            let mut wrong = body.clone();
+            wrong["output_bucket"] = json!("third");
+            match fixture.issue(&record, &wrong).await {
+                Err(error) => assert_eq!(error.code(), &S3ErrorCode::AccessDenied),
+                Ok(_) => panic!("A03_CROSS_BUCKET_OUTPUT_BOUND"),
+            }
+            let response = fixture.issue(&record, &body).await.expect("actual v2 issuance");
+            assert_eq!(response.output.0, StatusCode::OK);
+            assert_eq!(response.headers[http::header::CACHE_CONTROL], "no-store");
+            let issued: Value =
+                serde_json::from_slice(&response.output.1.collect().await.expect("issued body").to_bytes()).expect("issued JSON");
+            let (credential, root) = check_key_valid_with_context(
+                issued["session_token"].as_str().expect("token"),
+                issued["access_key_id"].as_str().expect("access key"),
+                Some(fixture.context.as_ref()),
+            )
+            .await
+            .expect("persisted actual temporary identity");
+            assert!(!root && credential.is_temp());
+            let output = format!("{prefix}image.png");
+            for (action, bucket, object, version, expected) in [
+                (
+                    S3Action::GetObjectVersionAction,
+                    "photos",
+                    "input/photo.png",
+                    Some(fixture.input_version.as_str()),
+                    true,
+                ),
+                (
+                    S3Action::GetObjectVersionAction,
+                    "photos",
+                    "input/photo.png",
+                    Some("wrong-version"),
+                    false,
+                ),
+                (S3Action::GetObjectAction, "photos", "input/photo.png", None, false),
+                (S3Action::PutObjectAction, "artifacts", output.as_str(), None, true),
+                (S3Action::PutObjectAction, "photos", output.as_str(), None, false),
+                (S3Action::PutObjectAction, "third", output.as_str(), None, false),
+                (S3Action::PutObjectAction, "artifacts", "output/other/image.png", None, false),
+            ] {
+                assert_eq!(
+                    fixture.allowed(&credential, action, bucket, object, version).await,
+                    expected,
+                    "A03_CROSS_BUCKET_PARENT_INTERSECTION: bucket={bucket} object={object}"
+                );
+            }
+            fixture.policy(PARENT, Fixture::parent_policy(true, true)).await;
+            assert!(
+                !fixture
+                    .allowed(&credential, S3Action::PutObjectAction, "artifacts", &output, None)
+                    .await,
+                "explicit parent Deny remains effective"
+            );
+            fixture.policy(PARENT, Fixture::parent_policy(false, false)).await;
+            assert!(
+                !fixture
+                    .allowed(&credential, S3Action::PutObjectAction, "artifacts", &output, None)
+                    .await,
+                "removing parent output permission remains effective"
+            );
+            assert!(
+                fixture
+                    .allowed(
+                        &credential,
+                        S3Action::GetObjectVersionAction,
+                        "photos",
+                        "input/photo.png",
+                        Some(&fixture.input_version)
+                    )
+                    .await
+            );
+            fixture
+                .iam
+                .revoke_plugin_delegation(PARENT, &record.id)
+                .await
+                .expect("revoke v2");
+            assert!(fixture.issue(&record, &body).await.is_err(), "revoked v2 cannot issue again");
+        },
+    )
+    .await;
+}

rustfs/src/admin/handlers/plugin_delegation.rs

@@ -0,0 +1,624 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+
+//! Restriction registration only. These endpoints do not vend credentials.
+use crate::admin::{
+    auth::{AdminResourceScope, validate_admin_action_with_bucket_object_for_iam},
+    plugin_runner::{self, ObjectReference, SubmitRequest},
+    router::{AdminOperation, Operation, S3Router},
+    runtime_sources,
+    storage_api::s3::{self, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result},
+};
+use crate::{
+    auth::{check_key_valid_with_context, get_session_token},
+    server::{ADMIN_PREFIX, RemoteAddr},
+};
+use http::{HeaderMap, HeaderValue, Method, StatusCode};
+use matchit::Params;
+use rustfs_iam::delegation::{DelegationError, Restrictions};
+use rustfs_policy::policy::action::{Action, AdminAction};
+use serde::Deserialize;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+pub fn register_plugin_delegation_route(router: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    router.insert(Method::POST, &format!("{ADMIN_PREFIX}/v3/plugin-delegations"), AdminOperation(&Create))?;
+    router.insert(Method::GET, &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}"), AdminOperation(&Get))?;
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}/revoke"),
+        AdminOperation(&Revoke),
+    )?;
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}/submit"),
+        AdminOperation(&Submit),
+    )?;
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}/runs/query"),
+        AdminOperation(&QueryRun),
+    )?;
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}/runs/history"),
+        AdminOperation(&HistoryRun),
+    )?;
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}/runs/cancel"),
+        AdminOperation(&CancelRun),
+    )?;
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}/runs/page"),
+        AdminOperation(&PageRuns),
+    )?;
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/plugin-delegations/{{id}}/runs/replay"),
+        AdminOperation(&ReplayRun),
+    )?;
+    Ok(())
+}
+
+pub(super) fn storage_error(error: DelegationError) -> S3Error {
+    match error {
+        DelegationError::Invalid => s3::error(S3ErrorCode::InvalidRequest, "invalid delegation restriction"),
+        DelegationError::NotFound | DelegationError::OwnerRejected => s3::error(S3ErrorCode::NoSuchKey, "delegation not found"),
+        DelegationError::Unavailable => s3::error(S3ErrorCode::InternalError, "delegation unavailable"),
+        _ => s3::error(S3ErrorCode::AccessDenied, "delegation rejected"),
+    }
+}
+
+pub(super) async fn owner(
+    req: &S3Request<Body>,
+    action: AdminAction,
+) -> S3Result<(std::sync::Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>, String)> {
+    let signed = req
+        .credentials
+        .as_ref()
+        .ok_or_else(|| s3::error(S3ErrorCode::AccessDenied, "authentication required"))?;
+    if get_session_token(&req.uri, &req.headers).is_some() {
+        return Err(s3::error(S3ErrorCode::AccessDenied, "original user required"));
+    }
+    let context = runtime_sources::app_context_from_req(req)
+        .ok_or_else(|| s3::error(S3ErrorCode::InternalError, "application context unavailable"))?;
+    let (credentials, root) = check_key_valid_with_context("", &signed.access_key, Some(context.as_ref())).await?;
+    if root || credentials.is_temp() || credentials.is_service_account() {
+        return Err(s3::error(S3ErrorCode::AccessDenied, "original user required"));
+    }
+    if !context.iam().is_ready() {
+        return Err(s3::error(S3ErrorCode::InternalError, "iam unavailable"));
+    }
+    let iam = context.iam().handle();
+    validate_admin_action_with_bucket_object_for_iam(
+        iam.clone(),
+        &req.headers,
+        &credentials,
+        false,
+        Action::AdminAction(action),
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|a| a.map(|a| a.0)),
+        AdminResourceScope::bucket_object("", ""),
+    )
+    .await?;
+    Ok((iam, credentials.access_key))
+}
+
+pub(super) fn response<T: serde::Serialize>(value: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let bytes =
+        serde_json::to_vec(value).map_err(|_| s3::error(S3ErrorCode::InternalError, "delegation response unavailable"))?;
+    let mut headers = HeaderMap::new();
+    headers.insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/json"));
+    headers.insert(http::header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
+    Ok(S3Response::with_headers((StatusCode::OK, Body::from(bytes)), headers))
+}
+
+struct Create;
+#[async_trait::async_trait]
+impl Operation for Create {
+    async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, owner) = owner(&req, AdminAction::CreatePluginDelegationAction).await?;
+        let bytes = tokio::time::timeout(Duration::from_secs(10), req.input.store_all_limited(16384))
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "delegation body unavailable"))?
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "delegation body unavailable"))?;
+        let restriction: Restrictions = serde_json::from_slice(&bytes)
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid delegation restriction"))?;
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_err(|_| s3::error(S3ErrorCode::InternalError, "clock unavailable"))?
+            .as_secs();
+        let record = iam
+            .create_plugin_delegation(&owner, restriction, now)
+            .await
+            .map_err(storage_error)?;
+        response(&record)
+    }
+}
+
+struct Get;
+#[async_trait::async_trait]
+impl Operation for Get {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, owner) = owner(&req, AdminAction::GetPluginDelegationAction).await?;
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_err(|_| s3::error(S3ErrorCode::InternalError, "clock unavailable"))?
+            .as_secs();
+        let record = iam
+            .read_active_plugin_delegation(&owner, id, now)
+            .await
+            .map_err(storage_error)?;
+        response(&record)
+    }
+}
+
+struct Revoke;
+#[async_trait::async_trait]
+impl Operation for Revoke {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, owner) = owner(&req, AdminAction::RevokePluginDelegationAction).await?;
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        iam.revoke_plugin_delegation(&owner, id).await.map_err(storage_error)?;
+        response(&serde_json::json!({"revoked":true,"id":id}))
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SubmitInput {
+    version: u32,
+    instance_id: String,
+    idempotency_key: String,
+    object: SubmitObject,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SubmitObject {
+    bucket: String,
+    key: String,
+    version_id: String,
+}
+
+struct Submit;
+#[async_trait::async_trait]
+impl Operation for Submit {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, signer) = owner(&req, AdminAction::SubmitPluginJobAction).await?;
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        let bytes = tokio::time::timeout(Duration::from_secs(10), req.input.store_all_limited(16384))
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "submit body unavailable"))?
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "submit body unavailable"))?;
+        let input: SubmitInput =
+            serde_json::from_slice(&bytes).map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid submit request"))?;
+        if input.version != 1
+            || !plugin_runner::literal(&input.instance_id)
+            || !plugin_runner::valid_idempotency_key(&input.idempotency_key)
+        {
+            return Err(s3::error(S3ErrorCode::InvalidRequest, "invalid submit request"));
+        }
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_err(|_| s3::error(S3ErrorCode::InternalError, "clock unavailable"))?
+            .as_secs();
+        let record = iam
+            .read_active_plugin_delegation(&signer, id, now)
+            .await
+            .map_err(storage_error)?;
+        let deployment = plugin_runner::configured_deployment()?;
+        if !plugin_runner::validate_delegation_scope(
+            &record,
+            &signer,
+            id,
+            &deployment,
+            ObjectReference {
+                bucket: &input.object.bucket,
+                key: &input.object.key,
+                version_id: &input.object.version_id,
+            },
+            now,
+        ) {
+            return Err(s3::error(S3ErrorCode::AccessDenied, "submit restriction rejected"));
+        }
+        let receipt = plugin_runner::submit(&SubmitRequest {
+            version: input.version,
+            instance_id: &input.instance_id,
+            idempotency_key: &input.idempotency_key,
+            object: ObjectReference {
+                bucket: &input.object.bucket,
+                key: &input.object.key,
+                version_id: &input.object.version_id,
+            },
+            delegation_id: &record.id,
+            delegation_revision: &record.revision,
+            actual_signer_access_key: &signer,
+        })
+        .await?;
+        response(&receipt)
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct QueryRunInput {
+    version: u32,
+    run_id: uuid::Uuid,
+}
+
+struct QueryRun;
+#[async_trait::async_trait]
+impl Operation for QueryRun {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, signer) = owner(&req, AdminAction::GetPluginRunAction).await?;
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        let bytes = tokio::time::timeout(Duration::from_secs(10), req.input.store_all_limited(16384))
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "query body unavailable"))?
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "query body unavailable"))?;
+        let input: QueryRunInput =
+            serde_json::from_slice(&bytes).map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid query request"))?;
+        if input.version != 1 || input.run_id.is_nil() {
+            return Err(s3::error(S3ErrorCode::InvalidRequest, "invalid query request"));
+        }
+        let record = iam.read_owned_plugin_delegation(&signer, id).await.map_err(storage_error)?;
+        let result = plugin_runner::query(&plugin_runner::QueryRequest {
+            version: input.version,
+            run_id: input.run_id,
+            delegation_id: &record.id,
+            delegation_revision: &record.revision,
+            actual_signer_access_key: &signer,
+        })
+        .await?;
+        response(&result)
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct HistoryRunInput {
+    version: u32,
+    run_id: uuid::Uuid,
+}
+
+fn valid_history_run_input(input: &HistoryRunInput) -> bool {
+    input.version == 1 && !input.run_id.is_nil()
+}
+
+struct HistoryRun;
+#[async_trait::async_trait]
+impl Operation for HistoryRun {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, signer) = owner(&req, AdminAction::ListPluginRunsAction).await?;
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        let bytes = tokio::time::timeout(Duration::from_secs(10), req.input.store_all_limited(16384))
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "history body unavailable"))?
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "history body unavailable"))?;
+        let input: HistoryRunInput =
+            serde_json::from_slice(&bytes).map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid history request"))?;
+        if !valid_history_run_input(&input) {
+            return Err(s3::error(S3ErrorCode::InvalidRequest, "invalid history request"));
+        }
+        let record = iam.read_owned_plugin_delegation(&signer, id).await.map_err(storage_error)?;
+        let result = plugin_runner::history(&plugin_runner::HistoryRequest {
+            version: input.version,
+            run_id: input.run_id,
+            delegation_id: &record.id,
+            delegation_revision: &record.revision,
+            actual_signer_access_key: &signer,
+        })
+        .await?;
+        response(&result)
+    }
+}
+
+struct CancelRun;
+#[async_trait::async_trait]
+impl Operation for CancelRun {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, signer) = owner(&req, AdminAction::CancelPluginRunAction).await?;
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        let bytes = tokio::time::timeout(Duration::from_secs(10), req.input.store_all_limited(16384))
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "cancel body unavailable"))?
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "cancel body unavailable"))?;
+        let input: QueryRunInput =
+            serde_json::from_slice(&bytes).map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid cancel request"))?;
+        if input.version != 1 || input.run_id.is_nil() {
+            return Err(s3::error(S3ErrorCode::InvalidRequest, "invalid cancel request"));
+        }
+        let record = iam.read_owned_plugin_delegation(&signer, id).await.map_err(storage_error)?;
+        let result = plugin_runner::cancel(&plugin_runner::CancelRequest {
+            version: input.version,
+            run_id: input.run_id,
+            delegation_id: &record.id,
+            delegation_revision: &record.revision,
+            actual_signer_access_key: &signer,
+        })
+        .await?;
+        response(&result)
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct PageRunsInput {
+    version: u32,
+    limit: u32,
+    #[serde(default)]
+    cursor: Option<plugin_runner::RunPageCursor>,
+}
+
+fn valid_page_runs_input(input: &PageRunsInput) -> bool {
+    input.version == 1 && (1..=100).contains(&input.limit) && !input.cursor.is_some_and(|cursor| cursor.run_id.is_nil())
+}
+
+struct PageRuns;
+#[async_trait::async_trait]
+impl Operation for PageRuns {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, signer) = owner(&req, AdminAction::ListPluginRunsAction).await?;
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        let bytes = tokio::time::timeout(Duration::from_secs(10), req.input.store_all_limited(16384))
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "page body unavailable"))?
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "page body unavailable"))?;
+        let input: PageRunsInput =
+            serde_json::from_slice(&bytes).map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid page request"))?;
+        if !valid_page_runs_input(&input) {
+            return Err(s3::error(S3ErrorCode::InvalidRequest, "invalid page request"));
+        }
+        let record = iam.read_owned_plugin_delegation(&signer, id).await.map_err(storage_error)?;
+        let result = plugin_runner::page(&plugin_runner::PageRequest {
+            version: input.version,
+            actual_signer_access_key: &signer,
+            delegation_id: &record.id,
+            delegation_revision: &record.revision,
+            limit: input.limit,
+            cursor: input.cursor,
+        })
+        .await?;
+        response(&result)
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ReplayRunInput {
+    version: u32,
+    origin_run_id: uuid::Uuid,
+    idempotency_key: String,
+    deadline_unix_micros: i64,
+}
+
+fn valid_replay_run_input(input: &ReplayRunInput) -> bool {
+    input.version == 1
+        && !input.origin_run_id.is_nil()
+        && plugin_runner::valid_idempotency_key(&input.idempotency_key)
+        && input.deadline_unix_micros > 0
+}
+
+struct ReplayRun;
+#[async_trait::async_trait]
+impl Operation for ReplayRun {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let (iam, signer) = owner(&req, AdminAction::ReplayPluginRunAction).await?;
+        let id = params
+            .get("id")
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "delegation id required"))?;
+        let bytes = tokio::time::timeout(Duration::from_secs(10), req.input.store_all_limited(16384))
+            .await
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "replay body unavailable"))?
+            .map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "replay body unavailable"))?;
+        let input: ReplayRunInput =
+            serde_json::from_slice(&bytes).map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "invalid replay request"))?;
+        if !valid_replay_run_input(&input) {
+            return Err(s3::error(S3ErrorCode::InvalidRequest, "invalid replay request"));
+        }
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_err(|_| s3::error(S3ErrorCode::InternalError, "clock unavailable"))?
+            .as_secs();
+        let record = iam
+            .read_active_plugin_delegation(&signer, id, now)
+            .await
+            .map_err(storage_error)?;
+        let result = plugin_runner::replay(&plugin_runner::ReplayRequest {
+            version: input.version,
+            origin_run_id: input.origin_run_id,
+            idempotency_key: &input.idempotency_key,
+            deadline_unix_micros: input.deadline_unix_micros,
+            actual_signer_access_key: &signer,
+            delegation_id: &record.id,
+            delegation_revision: &record.revision,
+        })
+        .await?;
+        response(&result)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[tokio::test]
+    async fn delegation_mutations_require_authenticated_original_user() {
+        for operation in 0..8 {
+            let req = S3Request {
+                input: Body::from("{}".to_owned()),
+                method: Method::POST,
+                uri: http::Uri::from_static("/rustfs/admin/v3/plugin-delegations"),
+                headers: HeaderMap::new(),
+                extensions: http::Extensions::new(),
+                credentials: None,
+                region: None,
+                service: None,
+                trailing_headers: None,
+            };
+            let result = if operation == 0 {
+                Get.call(req, Params::new()).await
+            } else if operation == 1 {
+                Revoke.call(req, Params::new()).await
+            } else if operation == 2 {
+                Submit.call(req, Params::new()).await
+            } else if operation == 3 {
+                QueryRun.call(req, Params::new()).await
+            } else if operation == 4 {
+                HistoryRun.call(req, Params::new()).await
+            } else if operation == 5 {
+                CancelRun.call(req, Params::new()).await
+            } else if operation == 6 {
+                PageRuns.call(req, Params::new()).await
+            } else if operation == 7 {
+                ReplayRun.call(req, Params::new()).await
+            } else {
+                Create.call(req, Params::new()).await
+            };
+            let error = result.expect_err("unsigned mutations must not reach persistence");
+            assert_eq!(error.code(), &S3ErrorCode::AccessDenied);
+        }
+    }
+
+    #[test]
+    fn submit_dto_rejects_client_authority_fields_and_null_versions() {
+        assert!(serde_json::from_str::<SubmitInput>(
+            r#"{"version":1,"instance_id":"img","idempotency_key":"k","object":{"bucket":"b","key":"k","version_id":"v"},"actual_signer_access_key":"forged"}"#
+        )
+        .is_err());
+        let input: SubmitInput = serde_json::from_str(
+            r#"{"version":1,"instance_id":"img","idempotency_key":"k","object":{"bucket":"b","key":"a/x","version_id":"null"}}"#,
+        )
+        .unwrap();
+        let record = rustfs_iam::delegation::Delegation {
+            version: 1,
+            id: "a".repeat(32),
+            revision: "a".repeat(32),
+            owner_access_key: "owner".into(),
+            created_at_unix_seconds: 1,
+            restrictions: Restrictions {
+                version: None,
+                output_bucket: None,
+                deployment: "prod".into(),
+                bucket: "b".into(),
+                input_prefix: "a/".into(),
+                output_prefix: "out/".into(),
+                expires_at_unix_seconds: 100,
+            },
+        };
+        assert!(!plugin_runner::validate_delegation_scope(
+            &record,
+            "owner",
+            &record.id,
+            "prod",
+            ObjectReference {
+                bucket: &input.object.bucket,
+                key: &input.object.key,
+                version_id: &input.object.version_id,
+            },
+            2,
+        ));
+    }
+
+    #[test]
+    fn query_run_dto_rejects_client_authority_fields() {
+        let run_id = uuid::Uuid::from_u128(1);
+        let input: QueryRunInput =
+            serde_json::from_str(&format!(r#"{{"version":1,"run_id":"{run_id}"}}"#)).expect("valid run query");
+        assert_eq!(input.run_id, run_id);
+        assert!(
+            serde_json::from_str::<QueryRunInput>(&format!(
+                r#"{{"version":1,"run_id":"{run_id}","actual_signer_access_key":"forged"}}"#
+            ))
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn history_run_dto_is_strict_and_rejects_nil_run() {
+        let run_id = uuid::Uuid::from_u128(1);
+        let input: HistoryRunInput =
+            serde_json::from_str(&format!(r#"{{"version":1,"run_id":"{run_id}"}}"#)).expect("valid history request");
+        assert!(valid_history_run_input(&input));
+        for raw in [
+            r#"{"version":1,"run_id":"00000000-0000-0000-0000-000000000000"}"#,
+            r#"{"version":2,"run_id":"00000000-0000-0000-0000-000000000001"}"#,
+            r#"{"version":1,"run_id":"00000000-0000-0000-0000-000000000001","actual_signer_access_key":"forged"}"#,
+        ] {
+            let rejected = match serde_json::from_str::<HistoryRunInput>(raw) {
+                Ok(input) => !valid_history_run_input(&input),
+                Err(_) => true,
+            };
+            assert!(rejected, "accepted invalid history request: {raw}");
+        }
+    }
+
+    #[test]
+    fn page_runs_dto_is_strict_and_bounded() {
+        let cursor = plugin_runner::RunPageCursor {
+            created_at_unix_micros: 10,
+            run_id: uuid::Uuid::from_u128(1),
+        };
+        let input: PageRunsInput = serde_json::from_str(
+            r#"{"version":1,"limit":100,"cursor":{"created_at_unix_micros":10,"run_id":"00000000-0000-0000-0000-000000000001"}}"#,
+        )
+        .expect("valid page request");
+        assert_eq!(input.cursor, Some(cursor));
+        for raw in [
+            r#"{"version":1,"limit":0}"#,
+            r#"{"version":1,"limit":101}"#,
+            r#"{"version":1,"limit":1,"unknown":true}"#,
+            r#"{"version":1,"limit":1,"actual_signer_access_key":"forged"}"#,
+            r#"{"version":1,"limit":1,"cursor":{"created_at_unix_micros":10,"run_id":"00000000-0000-0000-0000-000000000000"}}"#,
+        ] {
+            let rejected = match serde_json::from_str::<PageRunsInput>(raw) {
+                Ok(input) => !valid_page_runs_input(&input),
+                Err(_) => true,
+            };
+            assert!(rejected, "accepted invalid page request: {raw}");
+        }
+    }
+
+    #[test]
+    fn replay_run_dto_is_strict_and_bounded() {
+        let run_id = uuid::Uuid::from_u128(1);
+        let input: ReplayRunInput = serde_json::from_str(&format!(
+            r#"{{"version":1,"origin_run_id":"{run_id}","idempotency_key":"retry-1","deadline_unix_micros":123}}"#
+        ))
+        .expect("valid replay request");
+        assert!(valid_replay_run_input(&input));
+        for raw in [
+            r#"{"version":1,"origin_run_id":"00000000-0000-0000-0000-000000000000","idempotency_key":"k","deadline_unix_micros":123}"#.to_string(),
+            format!(r#"{{"version":1,"origin_run_id":"{run_id}","idempotency_key":"","deadline_unix_micros":123}}"#),
+            format!(r#"{{"version":1,"origin_run_id":"{run_id}","idempotency_key":"k","deadline_unix_micros":0}}"#),
+            format!(
+                r#"{{"version":1,"origin_run_id":"{run_id}","idempotency_key":"k","deadline_unix_micros":123,"actual_signer_access_key":"forged"}}"#
+            ),
+            format!(
+                r#"{{"version":1,"origin_run_id":"{run_id}","idempotency_key":"{}","deadline_unix_micros":123}}"#,
+                "x".repeat(1025)
+            ),
+        ] {
+            let rejected = match serde_json::from_str::<ReplayRunInput>(&raw) {
+                Ok(input) => !valid_replay_run_input(&input),
+                Err(_) => true,
+            };
+            assert!(rejected, "accepted invalid replay request: {raw}");
+        }
+    }
+}

rustfs/src/admin/mod.rs

@@ -17,6 +17,7 @@ mod auth;
 pub mod console;
 pub mod handlers;
 mod plugin_contract;
+pub(crate) mod plugin_runner;
 pub(crate) mod replication_metrics_wire;
 // Contract inventory is validated by tests before later runtime integration.
 pub(crate) mod route_policy;
@@ -64,6 +65,8 @@ pub fn make_admin_route(
 fn register_admin_routes(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
     health::register_health_route(r)?;
     sts::register_admin_auth_route(r)?;
+    handlers::plugin_delegation::register_plugin_delegation_route(r)?;
+    handlers::plugin_attempt::register_plugin_attempt_route(r)?;
 
     account::register_account_route(r)?;
     mfa::register_mfa_route(r)?;

rustfs/src/admin/route_policy.rs

@@ -25,6 +25,15 @@ const TLS_CERTIFICATE_PUBLISH: AdminActionRef = AdminActionRef::new("TLSCertific
 const TLS_CERTIFICATE_STATUS: AdminActionRef = AdminActionRef::new("TLSCertificateStatusAdminAction");
 const CONSOLE_LOG: AdminActionRef = AdminActionRef::new("ConsoleLogAdminAction");
 const COMMIT_TABLE: AdminActionRef = AdminActionRef::new("CommitTableAction");
+const ISSUE_PLUGIN_ATTEMPT: AdminActionRef = AdminActionRef::new("IssuePluginAttemptAction");
+const SUBMIT_PLUGIN_JOB: AdminActionRef = AdminActionRef::new("SubmitPluginJobAction");
+const GET_PLUGIN_RUN: AdminActionRef = AdminActionRef::new("GetPluginRunAction");
+const CANCEL_PLUGIN_RUN: AdminActionRef = AdminActionRef::new("CancelPluginRunAction");
+const LIST_PLUGIN_RUNS: AdminActionRef = AdminActionRef::new("ListPluginRunsAction");
+const REPLAY_PLUGIN_RUN: AdminActionRef = AdminActionRef::new("ReplayPluginRunAction");
+const GET_PLUGIN_DELEGATION: AdminActionRef = AdminActionRef::new("GetPluginDelegationAction");
+const CREATE_PLUGIN_DELEGATION: AdminActionRef = AdminActionRef::new("CreatePluginDelegationAction");
+const REVOKE_PLUGIN_DELEGATION: AdminActionRef = AdminActionRef::new("RevokePluginDelegationAction");
 const CREATE_POLICY: AdminActionRef = AdminActionRef::new("CreatePolicyAdminAction");
 const CREATE_USER: AdminActionRef = AdminActionRef::new("CreateUserAdminAction");
 const CREATE_SERVICE_ACCOUNT: AdminActionRef = AdminActionRef::new("CreateServiceAccountAdminAction");
@@ -656,6 +665,66 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         SERVER_INFO,
         RouteRiskLevel::Sensitive,
     ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations/{id}/issue-attempt",
+        ISSUE_PLUGIN_ATTEMPT,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations/{id}/submit",
+        SUBMIT_PLUGIN_JOB,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations/{id}/runs/query",
+        GET_PLUGIN_RUN,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations/{id}/runs/history",
+        LIST_PLUGIN_RUNS,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations/{id}/runs/cancel",
+        CANCEL_PLUGIN_RUN,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations/{id}/runs/page",
+        LIST_PLUGIN_RUNS,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations/{id}/runs/replay",
+        REPLAY_PLUGIN_RUN,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/plugin-delegations/{id}",
+        GET_PLUGIN_DELEGATION,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations",
+        CREATE_PLUGIN_DELEGATION,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/plugin-delegations/{id}/revoke",
+        REVOKE_PLUGIN_DELEGATION,
+        RouteRiskLevel::High,
+    ),
     admin(
         HttpMethod::Get,
         "/rustfs/admin/v4/plugins/catalog",
@@ -1933,6 +2002,72 @@ mod tests {
         assert!(validate_admin_route_policy_specs().is_ok());
     }
 
+    #[test]
+    fn plugin_submit_route_is_registered_as_high_risk_authenticated_action() {
+        let path = "/rustfs/admin/v3/plugin-delegations/{id}/submit";
+        assert_action(HttpMethod::Post, path, SUBMIT_PLUGIN_JOB);
+        let spec = ADMIN_ROUTE_POLICY_SPECS
+            .iter()
+            .find(|spec| spec.method() == HttpMethod::Post && spec.path() == path)
+            .expect("submit route policy");
+        assert_eq!(spec.risk_level(), RouteRiskLevel::High);
+    }
+
+    #[test]
+    fn plugin_run_query_route_uses_dedicated_authenticated_action() {
+        let path = "/rustfs/admin/v3/plugin-delegations/{id}/runs/query";
+        assert_action(HttpMethod::Post, path, GET_PLUGIN_RUN);
+        let spec = ADMIN_ROUTE_POLICY_SPECS
+            .iter()
+            .find(|spec| spec.method() == HttpMethod::Post && spec.path() == path)
+            .expect("plugin run query route policy");
+        assert_eq!(spec.risk_level(), RouteRiskLevel::Sensitive);
+    }
+
+    #[test]
+    fn plugin_run_history_route_uses_list_action() {
+        let path = "/rustfs/admin/v3/plugin-delegations/{id}/runs/history";
+        assert_action(HttpMethod::Post, path, LIST_PLUGIN_RUNS);
+        let spec = ADMIN_ROUTE_POLICY_SPECS
+            .iter()
+            .find(|spec| spec.method() == HttpMethod::Post && spec.path() == path)
+            .expect("plugin run history route policy");
+        assert_eq!(spec.risk_level(), RouteRiskLevel::Sensitive);
+    }
+
+    #[test]
+    fn plugin_run_cancel_route_uses_dedicated_authenticated_action() {
+        let path = "/rustfs/admin/v3/plugin-delegations/{id}/runs/cancel";
+        assert_action(HttpMethod::Post, path, CANCEL_PLUGIN_RUN);
+        let spec = ADMIN_ROUTE_POLICY_SPECS
+            .iter()
+            .find(|spec| spec.method() == HttpMethod::Post && spec.path() == path)
+            .expect("plugin run cancel route policy");
+        assert_eq!(spec.risk_level(), RouteRiskLevel::Sensitive);
+    }
+
+    #[test]
+    fn plugin_run_page_route_uses_dedicated_authenticated_action() {
+        let path = "/rustfs/admin/v3/plugin-delegations/{id}/runs/page";
+        assert_action(HttpMethod::Post, path, LIST_PLUGIN_RUNS);
+        let spec = ADMIN_ROUTE_POLICY_SPECS
+            .iter()
+            .find(|spec| spec.method() == HttpMethod::Post && spec.path() == path)
+            .expect("plugin run page route policy");
+        assert_eq!(spec.risk_level(), RouteRiskLevel::Sensitive);
+    }
+
+    #[test]
+    fn plugin_run_replay_route_uses_dedicated_authenticated_action() {
+        let path = "/rustfs/admin/v3/plugin-delegations/{id}/runs/replay";
+        assert_action(HttpMethod::Post, path, REPLAY_PLUGIN_RUN);
+        let spec = ADMIN_ROUTE_POLICY_SPECS
+            .iter()
+            .find(|spec| spec.method() == HttpMethod::Post && spec.path() == path)
+            .expect("plugin run replay route policy");
+        assert_eq!(spec.risk_level(), RouteRiskLevel::Sensitive);
+    }
+
     #[test]
     fn route_policy_inventory_has_no_internal_overlap() {
         let mut seen = BTreeSet::new();

rustfs/src/admin/route_registration_test.rs

@@ -116,6 +116,48 @@ fn registered_admin_router() -> S3Router<AdminOperation> {
 
 fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
     vec![
+        admin_route(Method::POST, "/v3/plugin-delegations"),
+        admin_route_sample(Method::GET, "/v3/plugin-delegations/{id}", "/v3/plugin-delegations/test-delegation"),
+        admin_route_sample(
+            Method::POST,
+            "/v3/plugin-delegations/{id}/revoke",
+            "/v3/plugin-delegations/test-delegation/revoke",
+        ),
+        admin_route_sample(
+            Method::POST,
+            "/v3/plugin-delegations/{id}/submit",
+            "/v3/plugin-delegations/test-delegation/submit",
+        ),
+        admin_route_sample(
+            Method::POST,
+            "/v3/plugin-delegations/{id}/runs/query",
+            "/v3/plugin-delegations/test-delegation/runs/query",
+        ),
+        admin_route_sample(
+            Method::POST,
+            "/v3/plugin-delegations/{id}/runs/history",
+            "/v3/plugin-delegations/test-delegation/runs/history",
+        ),
+        admin_route_sample(
+            Method::POST,
+            "/v3/plugin-delegations/{id}/runs/cancel",
+            "/v3/plugin-delegations/test-delegation/runs/cancel",
+        ),
+        admin_route_sample(
+            Method::POST,
+            "/v3/plugin-delegations/{id}/runs/page",
+            "/v3/plugin-delegations/test-delegation/runs/page",
+        ),
+        admin_route_sample(
+            Method::POST,
+            "/v3/plugin-delegations/{id}/runs/replay",
+            "/v3/plugin-delegations/test-delegation/runs/replay",
+        ),
+        admin_route_sample(
+            Method::POST,
+            "/v3/plugin-delegations/{id}/issue-attempt",
+            "/v3/plugin-delegations/test-delegation/issue-attempt",
+        ),
         route(Method::GET, HEALTH_PREFIX),
         route(Method::HEAD, HEALTH_PREFIX),
         route(Method::GET, HEALTH_READY_PATH),

rustfs/src/startup_background.rs

@@ -81,6 +81,14 @@ pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Resu
         let admission = None;
         let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone()).with_admission(admission.clone()));
         init_heal_manager_with_workload_provider(heal_storage, None, Some(workload_provider.clone())).await?;
+        if let Some(admission) = &admission
+            && !rustfs_scanner::set_scanner_node_admission(admission.clone())
+        {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::AlreadyExists,
+                "scanner node admission is already initialized",
+            ));
+        }
         #[cfg(any(target_os = "linux", all(test, unix)))]
         if let Some(admission) = admission {
             crate::background_admission::start(admission, background_cancel.clone(), workload_provider);

rustfs/src/storage/access.rs

@@ -64,6 +64,13 @@ use std::sync::OnceLock;
 use time::OffsetDateTime;
 use url::{Url, form_urlencoded};
 
+fn iam_authorization_error(err: IamError) -> ApiError {
+    match err {
+        IamError::IAMActionNotAllowed => ApiError::access_denied(),
+        other => ApiError::from(other),
+    }
+}
+
 const EVENT_OBJECT_TAG_AUTHORIZATION: &str = "object_tag_authorization";
 const LOG_COMPONENT_ACCESS: &str = "storage_access";
 const LOG_SUBSYSTEM_AUTHORIZATION: &str = "authorization";
@@ -1483,7 +1490,7 @@ async fn authorize_request_with_location_constraint<T>(
             let allowed = iam_store
                 .try_eval_prepared(&prepared, &final_args)
                 .await
-                .map_err(ApiError::from)?;
+                .map_err(iam_authorization_error)?;
             if !allowed
                 && matches!(
                     action,
@@ -1505,7 +1512,7 @@ async fn authorize_request_with_location_constraint<T>(
                 if !iam_store
                     .try_eval_prepared(&prepared, &final_args)
                     .await
-                    .map_err(ApiError::from)?
+                    .map_err(iam_authorization_error)?
                 {
                     return Err(denial.deny("iam_explicit_deny", action));
                 }
@@ -3618,6 +3625,14 @@ mod tests {
     use std::{collections::HashMap, sync::Arc};
     use time::OffsetDateTime;
 
+    #[test]
+    fn terminal_iam_denial_is_access_denied_and_backend_failure_stays_internal() {
+        let denied = s3s::S3Error::from(super::iam_authorization_error(rustfs_iam::error::Error::IAMActionNotAllowed));
+        assert_eq!(denied.code(), &S3ErrorCode::AccessDenied);
+        let unavailable = s3s::S3Error::from(super::iam_authorization_error(rustfs_iam::error::Error::IamSysNotInitialized));
+        assert_eq!(unavailable.code(), &S3ErrorCode::InternalError);
+    }
+
     struct UnreadyIam;
 
     impl IamInterface for UnreadyIam {

rustfs/src/storage/scanner_dirty_journal.rs

@@ -384,6 +384,10 @@ fn decode_owned_replay_record(owner: &str, bytes: &[u8]) -> Option<(JournalState
 
 fn encode_owned_replay_record(owner: &str, state: &JournalState) -> Result<Vec<u8>, serde_json::Error> {
     let mut entries = state.replay_entries();
+    let mut prefix = Vec::with_capacity(owner.len() + 32);
+    prefix.extend_from_slice(b"{\"owner\":");
+    serde_json::to_writer(&mut prefix, owner)?;
+    prefix.extend_from_slice(b",\"replay\":");
     for compact in [false, true] {
         if let Some(entries) = &mut entries {
             if compact {
@@ -391,13 +395,12 @@ fn encode_owned_replay_record(owner: &str, state: &JournalState) -> Result<Vec<u
                     entry.scope = ScannerDurableDirtyUsageReplayScope::WholeBucket;
                 }
             }
-            if let Ok(replay) = rustfs_scanner::encode_durable_dirty_usage_producer_replay_record(entries.clone())
-                && let Ok(replay) = serde_json::from_slice(&replay)
-            {
-                let bytes = serde_json::to_vec(&OwnedReplayRecord {
-                    owner: owner.to_string(),
-                    replay: Some(replay),
-                })?;
+            if let Ok(replay) = rustfs_scanner::encode_durable_dirty_usage_producer_replay_record(entries.clone()) {
+                // This producer already validated and serialized the complete frame.
+                let mut bytes = prefix.clone();
+                bytes.reserve(replay.len() + 1);
+                bytes.extend_from_slice(&replay);
+                bytes.push(b'}');
                 if bytes.len() <= DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES {
                     return Ok(bytes);
                 }
@@ -405,10 +408,8 @@ fn encode_owned_replay_record(owner: &str, state: &JournalState) -> Result<Vec<u
         }
     }
     // Never label a truncated record as complete producer coverage.
-    serde_json::to_vec(&OwnedReplayRecord {
-        owner: owner.to_string(),
-        replay: None,
-    })
+    prefix.extend_from_slice(b"null}");
+    Ok(prefix)
 }
 
 async fn run_durable_dirty_usage_journal(store: Arc<ECStore>, owner: String, shared: Weak<SharedJournal>, changed: Arc<Notify>) {
@@ -575,6 +576,63 @@ mod tests {
         ENV.get_or_init(|| rustfs_test_utils::TestECStoreEnv::builder().build()).await
     }
 
+    fn legacy_owned_encoding(owner: &str, state: &JournalState) -> Vec<u8> {
+        let mut entries = state.replay_entries();
+        for compact in [false, true] {
+            if let Some(entries) = &mut entries {
+                if compact {
+                    for entry in entries.iter_mut() {
+                        entry.scope = ScannerDurableDirtyUsageReplayScope::WholeBucket;
+                    }
+                }
+                if let Ok(replay) = rustfs_scanner::encode_durable_dirty_usage_producer_replay_record(entries.clone())
+                    && let Ok(replay) = serde_json::from_slice(&replay)
+                {
+                    let bytes = serde_json::to_vec(&OwnedReplayRecord {
+                        owner: owner.to_string(),
+                        replay: Some(replay),
+                    })
+                    .expect("legacy typed wrapper");
+                    if bytes.len() <= DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES {
+                        return bytes;
+                    }
+                }
+            }
+        }
+        serde_json::to_vec(&OwnedReplayRecord {
+            owner: owner.to_string(),
+            replay: None,
+        })
+        .expect("legacy unknown marker")
+    }
+
+    #[test]
+    fn owned_encoding_reuses_the_producer_frame_without_changing_wire_bytes() {
+        let mut scoped = JournalState::default();
+        for producer in SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION {
+            scoped.record_mutation("bucket-a".to_string(), "quote\"雪/object", 17, producer);
+        }
+        scoped.record_mutation("bucket-b".to_string(), "object", 9, SegmentInvalidationProducerIdentity::PutObject);
+        let unknown = JournalState {
+            invalidated: true,
+            ..Default::default()
+        };
+        for owner in ["owner", "quote\"slash\\newline\n", "所有者", "\0"] {
+            for state in [&scoped, &unknown, &JournalState::default()] {
+                let expected = legacy_owned_encoding(owner, state);
+                let actual = encode_owned_replay_record(owner, state).expect("complete owned encoding");
+                assert_eq!(actual, expected, "legacy bytes for owner {owner:?}");
+            }
+        }
+    }
+
+    #[test]
+    fn owned_unknown_marker_keeps_the_pinned_legacy_wire_fixture() {
+        let bytes = encode_owned_replay_record("owner", &JournalState::default()).expect("unknown marker");
+        assert_eq!(bytes, br#"{"owner":"owner","replay":null}"#);
+        assert!(decode_owned_replay_record("owner", &bytes).is_none());
+    }
+
     #[test]
     fn journal_owner_is_stable_and_never_defaults_to_a_shared_identity() {
         assert!(durable_dirty_usage_owner("").is_none());
@@ -654,6 +712,7 @@ mod tests {
         }
         assert!(state.scope_bytes <= DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES);
         let bytes = encode_owned_replay_record(&owner, &state).expect("bounded record");
+        assert_eq!(bytes, legacy_owned_encoding(&owner, &state), "legacy compacted bytes");
         assert!(bytes.len() <= DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES);
         let (decoded, _) = decode_owned_replay_record(&owner, &bytes).expect("compacted replay");
         assert_eq!(decoded.buckets.len(), 3);
@@ -673,6 +732,7 @@ mod tests {
         assert!(!state.invalidated);
         assert_eq!(state.buckets.len(), 1024);
         let bytes = encode_owned_replay_record(&owner, &state).expect("unverified marker");
+        assert_eq!(bytes, legacy_owned_encoding(&owner, &state), "legacy unknown marker bytes");
         assert!(bytes.len() <= DURABLE_DIRTY_USAGE_REPLAY_MAX_BYTES);
         let record: OwnedReplayRecord = serde_json::from_slice(&bytes).expect("marker");
         assert!(record.replay.is_none());

scripts/scanner_process_receipt.py

@@ -18,6 +18,7 @@ SAFE_ENVIRONMENT = frozenset({
     "RUSTFS_STORAGE_CLASS_STANDARD", "RUSTFS_DURABILITY_MODE", "RUSTFS_NEW_BUCKET_DURABILITY_MODE",
     "RUSTFS_SCANNER_ENABLED", "RUSTFS_HEAL_ENABLED", "RUSTFS_SCANNER_SPEED", "RUSTFS_SCANNER_IDLE_MODE",
     "RUSTFS_SCANNER_START_DELAY_SECS", "RUSTFS_SCANNER_CYCLE",
+    "RUSTFS_HEAL_ADAPTIVE_ADMISSION",
     "RUSTFS_OBJECT_LOCK_DIAG_ENABLE", "RUSTFS_OBS_LOGS_EXPORT_ENABLED", "RUSTFS_OBS_TRACES_EXPORT_ENABLED",
     "RUSTFS_OBS_METRICS_EXPORT_ENABLED", "RUST_LOG", "NO_PROXY", "no_proxy",
 })

scripts/test_scanner_process_receipt.py

@@ -108,6 +108,21 @@ class ProcessReceiptTest(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "not active"):
             self.attest()
 
+    def test_adaptive_admission_is_bound_to_actual_process_environment(self):
+        key = "RUSTFS_HEAL_ADAPTIVE_ADMISSION"
+        self.public[key] = "true"
+        raw = self.environment + b"RUSTFS_HEAL_ADAPTIVE_ADMISSION=true\0"
+        (self.pid / "environ").write_bytes(raw)
+        result = self.attest()
+        self.assertEqual(result["environment_sha256"], receipt.fingerprint(self.public))
+        self.assertIn(key, result["environment_keys"])
+        for observed in (self.environment, raw.replace(b"ADMISSION=true", b"ADMISSION=false"),
+                         raw + b"RUSTFS_HEAL_ADAPTIVE_ADMISSION=true\0"):
+            with self.subTest(raw_hash=hashlib.sha256(observed).hexdigest()), self.assertRaises(ValueError):
+                receipt.environment_digest(observed, self.public)
+        with self.assertRaisesRegex(ValueError, "mismatch"):
+            receipt.environment_digest(raw, {**self.public, key: "false"})
+
     def test_wrong_binary_content_is_rejected(self):
         self.binary.write_bytes(b"replaced executable")
         with self.assertRaisesRegex(ValueError, "binary digest"):