Merge branch 'main' of https://github.com/rustfs/rustfs into fix/connect-consumer-fixtures-log

CHANGELOG.md

@@ -18,7 +18,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - Read paths: an object at or below `policy.inline_max_bytes` (16 MiB by default) is teed to the client and to the local store in a single source read; a larger object or a Range read streams through and a background pull stores the whole object. A HEAD miss is proxied to the source and stores nothing (`policy.head = local_only` disables it). Every source-backed response carries `x-rustfs-on-demand-migration: source`
   - Protections: a per-source circuit breaker, a per-key negative cache, singleflight per key, a concurrency limit and a bounded pull queue shared by the inline and background paths, an optional bandwidth limit, an anti-loop request marker, and the shared outbound-endpoint (SSRF) policy
   - Metrics under `rustfs_on_demand_migration_*` (`requests_total`, `pulled_bytes_total`, `pulled_objects_total`, `pull_failures_total`, `inflight_pulls`, `queue_depth`, `source_latency_seconds_*`, `breaker_state`), mirrored per node by the admin status route
-  - Limitations: listings show only local objects (the source is not merged into `ListObjectsV2`); PUT and DELETE never reach the source; a source object updated after it was pulled is not re-fetched; SSE-C source objects are unsupported and answer 424; `Last-Modified` on a pulled object is the local write time, with the source timestamp kept in metadata
+  - Listings: `ListObjects` v1 remains local with ordinary key markers. `ListObjectsV2` can merge source objects when `policy.list_through = true`; this is off by default
+  - Upgrade and rollback: finish upgrading every node before enabling ODM. An rc.5 node that writes bucket configuration drops the ODM fields from metadata; neither a later restart nor moving the service out of ECStore recovers them. Before rollback, disable ODM and securely retain the original full configuration and credentials. After every node returns to a compatible version, restore and validate that configuration. Redacted exports cannot replace the credential backup; source-only objects are unavailable through RustFS while ODM is disabled. See the upgrade and rollback section of `docs/operations/on-demand-migration.md`
+  - Optional Google dependencies: default and `full` server builds retain native GCS support. `cargo build -p rustfs --no-default-features --features ftps,webdav` excludes Google SDKs while preserving configuration decoding and redaction; native GCS ODM and tier operations require the `gcs` feature. Do not use that build with existing GCS-tiered data
+  - Limitations: PUT and DELETE never reach the source; a source object updated after it was pulled is not re-fetched; SSE-C source objects are unsupported and answer 424; `Last-Modified` on a pulled object is the local write time, with the source timestamp kept in metadata
 - **NATS JetStream Publish Path**: Opt-in at-least-once delivery for the NATS notify and audit targets. A NATS Core publish flushes to the connection without awaiting a broker acknowledgement, so an event can be lost across a broker restart or a reconnect after the send queue has already cleared it. A queued event now clears only after the JetStream `PublishAck`, so bucket notifications survive those interruptions. Off by default and byte-identical to the NATS Core path when disabled.
   - Three configuration keys per target: `JETSTREAM_ENABLE`, `JETSTREAM_STREAM_NAME`, and `JETSTREAM_ACK_TIMEOUT_SECS`, under the `RUSTFS_NOTIFY_NATS_` and `RUSTFS_AUDIT_NATS_` prefixes
   - Durable store-and-forward with a stable dedup id sent as the `Nats-Msg-Id` header, so a replay after a crash is collapsed by the server duplicate window

crates/ecstore/src/bucket/metadata_sys.rs

@@ -50,7 +50,7 @@ use uuid::Uuid;
 
 /// Opaque bucket configuration notifications for application-owned services.
 /// `None` withdraws a configuration; consumers validate nonempty bytes.
-pub type BucketConfigPublishHook = Box<dyn Fn(&str, &str, Option<(&[u8], OffsetDateTime)>) + Send + Sync>;
+pub type BucketConfigPublishHook = Box<dyn Fn(&str, &str, Option<(&[u8], OffsetDateTime, Uuid)>) + Send + Sync>;
 pub static BUCKET_CONFIG_PUBLISH_HOOK: std::sync::OnceLock<BucketConfigPublishHook> = std::sync::OnceLock::new();
 
 const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60);
@@ -405,7 +405,8 @@ fn sync_on_demand_migration(bucket: &str, bm: &BucketMetadata) {
         hook(
             bucket,
             super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG,
-            bm.on_demand_migration_config(),
+            bm.on_demand_migration_config()
+                .map(|(bytes, stamp)| (bytes, stamp, bm.bucket_incarnation_id)),
         );
     }
 }
@@ -4374,23 +4375,26 @@ mod tests {
 
     const ODM_JSON: &[u8] = br#"{"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#;
 
+    type RecordedOdmConfig = Option<(Vec<u8>, OffsetDateTime, Uuid)>;
+    type RecordedOdmHookCall = (String, RecordedOdmConfig);
+
     /// Every `(bucket, config)` the recording hook has seen. Tests filter by
     /// their own bucket name; the hook is process-wide and set once.
-    static ODM_HOOK_CALLS: std::sync::Mutex<Vec<(String, Option<(Vec<u8>, OffsetDateTime)>)>> = std::sync::Mutex::new(Vec::new());
+    static ODM_HOOK_CALLS: std::sync::Mutex<Vec<RecordedOdmHookCall>> = std::sync::Mutex::new(Vec::new());
 
     fn install_recording_odm_hook() {
         BUCKET_CONFIG_PUBLISH_HOOK.get_or_init(|| {
             Box::new(|bucket, config_file, config| {
                 assert_eq!(config_file, super::super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG);
-                ODM_HOOK_CALLS
+                ODM_HOOK_CALLS.lock().unwrap().push((
-                    .lock()
+                    bucket.to_string(),
-                    .unwrap()
+                    config.map(|(bytes, stamp, incarnation)| (bytes.to_vec(), stamp, incarnation)),
-                    .push((bucket.to_string(), config.map(|(bytes, stamp)| (bytes.to_vec(), stamp))));
+                ));
             })
         });
     }
 
-    fn odm_hook_calls(bucket: &str) -> Vec<Option<(Vec<u8>, OffsetDateTime)>> {
+    fn odm_hook_calls(bucket: &str) -> Vec<RecordedOdmConfig> {
         ODM_HOOK_CALLS
             .lock()
             .unwrap()
@@ -4414,18 +4418,21 @@ mod tests {
             std::fs::create_dir_all(dir.path().join(bucket)).expect("physical bucket should exist");
         }
 
+        let incarnation = Uuid::new_v4();
         let expect_publish = |before: usize, label: &str| {
             let calls = odm_hook_calls(bucket);
             assert_eq!(calls.len(), before + 1, "{label} must publish exactly once");
             assert_eq!(
-                calls.last().unwrap().as_ref().map(|(bytes, _)| bytes.as_slice()),
+                calls.last().unwrap().as_ref().map(|(bytes, _, _)| bytes.as_slice()),
                 Some(ODM_JSON),
                 "{label} must publish the stored bytes"
             );
+            assert_eq!(calls.last().unwrap().as_ref().map(|(_, _, id)| *id), Some(incarnation));
         };
 
         // set (via persist_new_and_set, which installs through `set`).
         let mut bm = BucketMetadata::new(bucket);
+        bm.bucket_incarnation_id = incarnation;
         bm.update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
             .unwrap();
         let writer = BucketMetadataSys::new(ecstore.clone());
@@ -4475,7 +4482,7 @@ mod tests {
         let calls = odm_hook_calls(bucket);
         assert_eq!(calls.len(), before + 1);
         assert_eq!(
-            calls.last().unwrap().as_ref().map(|(bytes, _)| bytes.as_slice()),
+            calls.last().unwrap().as_ref().map(|(bytes, _, _)| bytes.as_slice()),
             Some(b"not-json".as_slice()),
             "the application validates opaque config bytes"
         );

crates/ecstore/src/store/bucket.rs

@@ -329,11 +329,11 @@ impl ECStore {
     /// reuse its result, which is sound because bucket deletion/recreation
     /// requires the lifecycle WRITE lock and therefore cannot have run while
     /// any read guard was continuously held.
-    pub(crate) async fn acquire_bucket_incarnation_fence(
+    pub async fn acquire_bucket_incarnation_fence(
         &self,
         bucket: &str,
         expected: uuid::Uuid,
-    ) -> Result<super::bucket_fence::BucketIncarnationFenceGuard> {
+    ) -> Result<super::BucketIncarnationFenceGuard> {
         let inner = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
         let pieces = super::bucket_fence::FencePieces {
             registry: self.bucket_fence_registry.clone(),

crates/ecstore/src/store/bucket_fence.rs

@@ -150,7 +150,7 @@ impl BucketFenceRegistry {
 /// A held bucket lifecycle read lock plus its registration in the fence
 /// registry. Dropping the guard deregisters it; the memo is cleared when the
 /// last guard for the bucket drops (or a lost lock is observed).
-pub(crate) struct BucketIncarnationFenceGuard {
+pub struct BucketIncarnationFenceGuard {
     inner: Option<NamespaceLockGuard>,
     registry: Arc<BucketFenceRegistry>,
     bucket: String,
@@ -158,6 +158,14 @@ pub(crate) struct BucketIncarnationFenceGuard {
 }
 
 impl BucketIncarnationFenceGuard {
+    /// Propagate lifecycle lock loss into the storage commit checks.
+    /// The caller still owns this guard until the complete write tail drains.
+    pub fn attach_to_object_options(&self, opts: &mut crate::object_api::ObjectOptions) {
+        if let Some(guard) = self.namespace_lock_guard() {
+            opts.add_bucket_lifecycle_lock_guard(guard);
+        }
+    }
+
     pub(crate) fn is_lock_lost(&self) -> bool {
         self.inner.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost)
     }
@@ -346,6 +354,36 @@ mod tests {
         first_pieces.abandon("b", first.token);
     }
 
+    #[tokio::test]
+    async fn checkpoint_options_inherit_bucket_fence_lock_loss() {
+        let lock = NamespaceLock::new("bucket-fence-options".to_string(), Arc::new(LocalClient::new()));
+        let inner = lock
+            .acquire_guard(&lock_request("options"))
+            .await
+            .expect("acquire")
+            .expect("quorum");
+        let pieces = FencePieces {
+            registry: Arc::default(),
+            inner,
+        };
+        let registration = pieces.enter("b");
+        let fence = pieces.into_guard("b", registration.token);
+        let mut opts = crate::object_api::ObjectOptions::default();
+        fence.attach_to_object_options(&mut opts);
+        let inherited = opts
+            .bucket_lifecycle_lock_fence
+            .as_ref()
+            .expect("checkpoint inherits lifecycle guard");
+        assert!(!inherited.is_lock_lost());
+        tokio::time::timeout(
+            Duration::from_secs(2),
+            fence.namespace_lock_guard().expect("held guard").lock_lost_notified(),
+        )
+        .await
+        .expect("distributed guard expires");
+        assert!(inherited.is_lock_lost(), "the actual pre-rename options must observe lifecycle lock loss");
+    }
+
     #[test]
     fn buckets_are_isolated() {
         let reg = BucketFenceRegistry::default();

crates/ecstore/src/store/mod.rs

@@ -417,6 +417,7 @@ const MAX_UPLOADS_LIST: usize = 10000;
 mod bucket;
 mod bucket_fence;
 pub(crate) use bucket::await_bucket_namespace_operation;
+pub use bucket_fence::BucketIncarnationFenceGuard;
 mod heal;
 mod heal_walk;
 pub use heal_walk::HealWalkVersion;

docs/architecture/crate-boundaries.md

@@ -87,6 +87,12 @@ The guard requires the documents and section headings listed in its `require_sou
 
 ## On-Demand Migration Service
 
+Read-through, backfill and external pull orchestration belong in an application
+service under `rustfs/src/<service>/`. ECStore owns the storage primitives they
+need, including atomic commits, lifecycle locks and on-disk metadata. A service
+may use these primitives without moving its provider clients or scheduling
+policy into the engine.
+
 `rustfs/src/on_demand_migration/` owns source clients, pull scheduling, list
 merging, runtime state and backfill orchestration. Its `storage_api.rs` is the
 only ECStore facade boundary. Object write-back still enters the application's
@@ -101,6 +107,13 @@ deployment constraints in the admin use case before the incarnation-fenced
 metadata update. Backfill reads metadata from its store's instance context and
 preserves the checkpoint ETag compare-and-set, lease and tail-drained writes.
 
+An ODM runtime is bound to the bucket incarnation published with its metadata,
+not just its name. Source reads and write-back reject a different incarnation.
+Checkpoint writes hold the user bucket's lifecycle fence through their complete
+commit and read-back, even if their caller stops waiting; the storage commit
+also observes lock loss. Deleting and recreating a bucket must not let work for
+its previous incarnation repopulate objects or checkpoints.
+
 Observability owns its metric DTOs and accepts application snapshot callbacks;
 it does not depend on the ODM runtime. The application registers both bucket
 and backfill snapshots during startup, before metadata and metric collection.

docs/operations/bucket-metadata-recovery.md

@@ -12,7 +12,7 @@ To inspect readable configurations while identifying failures, use the same auth
 
 ## Recover unreadable replication targets
 
-MinIO target configuration may be an array or KMS-encrypted data that RustFS cannot decode. Diagnosis preserves the failure instead of interpreting it as an empty target set.
+RustFS currently accepts the documented `{"targets": [...]}` object format. It cannot decrypt MinIO KMS-encrypted target metadata. Unreadable target payloads remain failures instead of being interpreted as an empty target set; diagnostic export and replacement import do not add MinIO KMS decryption support.
 
 1. Inspect the diagnostic manifest to identify affected buckets. Preserve a separate backup of the original source configuration and any credentials needed for recovery.
 2. Prepare a ZIP containing `<bucket>/bucket-targets.json` with a valid RustFS replacement, whose top-level shape is `{"targets": [...]}`. Supply the intended target settings and credentials; exported credentials are redacted. Use `{"targets": []}` only when intentionally clearing all targets, and reconcile any replication rules that reference removed targets.

rustfs/src/app/bucket_list_through.rs

@@ -28,7 +28,7 @@ use super::storage_api::bucket_usecase::StorageObjectOptions;
 use super::storage_api::bucket_usecase::bucket::versioning_sys::BucketVersioningSys;
 use super::storage_api::bucket_usecase::contract::list::{ListObjectsV2Info as StorageListObjectsV2Info, ListOperations as _};
 use super::storage_api::bucket_usecase::contract::object::ObjectOperations as _;
-use super::storage_api::bucket_usecase::s3::{S3Error, S3ErrorCode, S3Result};
+use super::storage_api::bucket_usecase::s3::{S3Error, S3ErrorCode, S3Request, S3Result};
 use super::storage_api::bucket_usecase::s3_api::bucket::ListObjectsV2Params;
 use crate::app::object::shared::{odm_source_unavailable_error, odm_state_error_class};
 use crate::error::ApiError;
@@ -101,16 +101,38 @@ fn invalid_continuation_token(err: &ListThroughTokenError) -> S3Error {
 /// bucket has no source, `list_through` is off, or the request carries the
 /// `source-proxy-request` anti-loop marker and therefore comes from a peer
 /// that must be answered locally.
-pub(crate) fn list_through_state(bucket: &str, headers: &HeaderMap) -> Option<Arc<BucketOdmState>> {
+pub(crate) async fn list_through_state<T>(
-    if get_header(headers, SUFFIX_SOURCE_PROXY_REQUEST).is_some() {
+    store: &ECStore,
-        return None;
+    bucket: &str,
+    req: &S3Request<T>,
+    params: &ListObjectsV2Params,
+) -> S3Result<Option<Arc<BucketOdmState>>> {
+    if get_header(&req.headers, SUFFIX_SOURCE_PROXY_REQUEST).is_some() {
+        return Ok(None);
     }
     let sys = OnDemandMigrationSys::get();
     if !sys.is_module_enabled() {
-        return None;
+        return Ok(None);
     }
-    let state = sys.state(bucket)?;
+    let Some(state) = sys.state(bucket).filter(|state| state.config().policy.list_through) else {
-    state.config().policy.list_through.then_some(state)
+        return Ok(None);
+    };
+    if params.max_keys == 0
+        || matches!(
+            source_list_plan(&params.prefix, state.config().filter.prefix.as_deref(), params.delimiter.as_deref()),
+            SourceListPlan::Skip,
+        )
+    {
+        return Ok(None);
+    }
+    let Some(expected_incarnation) = super::storage_api::bucket_usecase::access::odm_read_generation(req, bucket)? else {
+        return Ok(None);
+    };
+    let incarnation = store.bucket_incarnation_id(bucket).await.map_err(ApiError::from)?;
+    if incarnation != expected_incarnation {
+        return Ok(None);
+    }
+    Ok(state.filter_incarnation(incarnation))
 }
 
 /// A merged page plus whether the source had to be left out of it.
@@ -444,10 +466,11 @@ mod tests {
     use crate::app::bucket_usecase::DefaultBucketUsecase;
     use crate::app::gating_test_env::{run_large_stack_test, shared_gating_ecstore};
     use crate::app::storage_api::bucket_usecase::s3::{
-        ListObjectsInput, ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response, XmlSerialize, XmlSerializer,
+        GetObjectInput, HeadObjectInput, ListObjectsInput, ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response,
+        XmlSerialize, XmlSerializer,
     };
     use crate::app::storage_api::test::StoragePutObjReader;
-    use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+    use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
     use crate::app::storage_api::test::contract::object::ObjectIO as _;
     use crate::on_demand_migration::{
         FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig,
@@ -727,7 +750,12 @@ mod tests {
                 ..Default::default()
             },
         };
-        sys.apply(&bucket, Some(&config)).await;
+        sys.apply_for_incarnation(
+            &bucket,
+            store.bucket_incarnation_id(&bucket).await.expect("bucket identity"),
+            Some(&config),
+        )
+        .await;
         assert!(
             sys.state(&bucket).expect("ODM state installed").client().is_ok(),
             "fake source client must build"
@@ -797,6 +825,128 @@ mod tests {
         (result, requests)
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn stale_bucket_state_cannot_send_get_head_or_list_to_the_source() {
+        run_large_stack_test("list-through-incarnation", || async {
+            temp_env::async_with_vars(
+                [
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    let (endpoint, server, stop) = list_source(std::iter::repeat(source_xml(None, false, Some("source")))).await;
+                    let (_guard, input) = source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, None).await;
+                    let sys = OnDemandMigrationSys::get();
+                    let old = sys.state(&input.bucket).expect("original source state");
+                    let store = shared_gating_ecstore().await;
+                    let get = S3Request {
+                        input: GetObjectInput {
+                            bucket: input.bucket.clone(),
+                            key: "missing".into(),
+                            ..Default::default()
+                        },
+                        method: http::Method::GET,
+                        uri: http::Uri::from_static("/missing"),
+                        headers: HeaderMap::new(),
+                        extensions: http::Extensions::new(),
+                        credentials: None,
+                        region: None,
+                        service: None,
+                        trailing_headers: None,
+                    };
+                    let authorized_generation =
+                        super::super::storage_api::bucket_usecase::access::load_bucket_generation_from_store(
+                            &store,
+                            &get,
+                            &input.bucket,
+                        )
+                        .await
+                        .expect("capture the identity before authorization");
+                    store
+                        .delete_bucket(
+                            &input.bucket,
+                            &DeleteBucketOptions {
+                                force: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("delete original bucket");
+                    store
+                        .make_bucket(&input.bucket, &MakeBucketOptions::default())
+                        .await
+                        .expect("recreate bucket");
+                    let replacement = store
+                        .bucket_incarnation_id(&input.bucket)
+                        .await
+                        .expect("replacement identity");
+                    assert_ne!(replacement, old.incarnation_id());
+                    sys.remove(&input.bucket);
+                    let mut without_source = get.clone();
+                    super::super::storage_api::bucket_usecase::access::prepare_odm_read_generation(
+                        &store,
+                        &mut without_source,
+                        &input.bucket,
+                    )
+                    .await;
+                    for state_incarnation in [old.incarnation_id(), replacement] {
+                        sys.apply_for_incarnation(&input.bucket, state_incarnation, Some(old.config()))
+                            .await;
+                        // Both a stale runtime and a newly published replacement must reject
+                        // requests already authorized for the deleted incarnation.
+                        for capture in 0..3 {
+                            if capture == 0 && state_incarnation == replacement {
+                                continue;
+                            }
+                            let mut get = if capture == 2 { without_source.clone() } else { get.clone() };
+                            if capture == 1 {
+                                get.extensions.insert(authorized_generation.clone());
+                            }
+                            let mut head = get.clone().map_input(|_| HeadObjectInput {
+                                bucket: input.bucket.clone(),
+                                key: "missing".into(),
+                                ..Default::default()
+                            });
+                            head.method = http::Method::HEAD;
+                            let mut list = get.clone().map_input(|_| input.clone());
+                            list.uri = http::Uri::from_static("/?list-type=2");
+                            let usecase = crate::app::object::DefaultObjectUsecase::from_global();
+                            let get_error = tokio::time::timeout(Duration::from_secs(10), usecase.execute_get_object(get))
+                                .await
+                                .expect("GET stays local")
+                                .expect_err("local object is absent");
+                            assert_eq!(*get_error.code(), S3ErrorCode::NoSuchKey);
+                            let head_error = tokio::time::timeout(Duration::from_secs(10), usecase.execute_head_object(head))
+                                .await
+                                .expect("HEAD stays local")
+                                .expect_err("local object is absent");
+                            assert_eq!(*head_error.code(), S3ErrorCode::NoSuchKey);
+                            let listing = tokio::time::timeout(
+                                Duration::from_secs(10),
+                                DefaultBucketUsecase::from_global().execute_list_objects_v2(list),
+                            )
+                            .await
+                            .expect("LIST stays local")
+                            .expect("replacement bucket lists locally");
+                            assert_eq!(listing.output.key_count, Some(0));
+                        }
+                    }
+                    stop.cancel();
+                    assert!(server.await.expect("source server must remain unused").is_empty());
+                },
+            )
+            .await;
+        });
+    }
+
     #[test]
     #[serial_test::serial]
     fn list_objects_v1_stays_local_with_xml_safe_key_markers() {

rustfs/src/app/bucket_usecase.rs

@@ -20,7 +20,7 @@ use super::storage_api::bucket_usecase::StorageObjectInfo as ObjectInfo;
 use super::storage_api::bucket_usecase::access::ReqInfo;
 use super::storage_api::bucket_usecase::access::{
     authorize_request, bucket_config_mutation_incarnation, log_list_buckets_iam_implicit_deny,
-    prepare_list_buckets_iam_authorization, req_info_ref,
+    prepare_list_buckets_iam_authorization, prepare_odm_read_generation, req_info_ref,
 };
 #[cfg(test)]
 use super::storage_api::bucket_usecase::bucket::target::BucketTarget;
@@ -2729,7 +2729,7 @@ impl DefaultBucketUsecase {
 
     async fn execute_list_objects_v2_inner(
         &self,
-        req: S3Request<ListObjectsV2Input>,
+        mut req: S3Request<ListObjectsV2Input>,
         allow_list_through: bool,
     ) -> S3Result<S3Response<ListObjectsV2Output>> {
         let ListObjectsV2Input {
@@ -2742,7 +2742,7 @@ impl DefaultBucketUsecase {
             prefix,
             start_after,
             ..
-        } = req.input;
+        } = req.input.clone();
 
         let params = parse_list_objects_v2_params(prefix, delimiter, max_keys, continuation_token, start_after)?;
 
@@ -2757,10 +2757,13 @@ impl DefaultBucketUsecase {
         // The on-demand migration envelope is decoded whether or not this
         // bucket still merges: a token handed out under `list_through` must keep
         // paginating after the policy is turned off (rustfs/backlog#2164).
+        if allow_list_through {
+            prepare_odm_read_generation(&store, &mut req, &bucket).await;
+        }
         let (merged_token, source_state) = if allow_list_through {
             (
                 list_through::decode_list_cursor(params.decoded_continuation_token.as_deref())?,
-                list_through::list_through_state(&bucket, &req.headers),
+                list_through::list_through_state(&store, &bucket, &req, &params).await?,
             )
         } else {
             (None, None)

rustfs/src/app/object/get.rs

@@ -3843,11 +3843,11 @@ impl DefaultObjectUsecase {
         if !odm_get_may_consult_source(opts, part_number) {
             return None;
         }
-        let lookup = OnDemandMigrationSys::get().resolve(bucket, key)?;
+        let sys = OnDemandMigrationSys::get();
-        let (state, client) = match odm_get_verdict(lookup) {
+        if !sys.is_module_enabled() {
-            OdmGetVerdict::Fail(err) => return Some(OdmGetOutcome::Respond(Err(err))),
+            return None;
-            OdmGetVerdict::Consult { state, client } => (state, client),
+        }
-        };
+        let state = sys.state(bucket).filter(|state| state.matches_prefix(key))?;
         let policy = &state.config().policy;
         // The read path reports a latest delete marker as a plain 404, so the
         // marker is classified here, and only where one can exist.
@@ -3861,6 +3861,24 @@ impl DefaultObjectUsecase {
                 None => return Some(OdmGetOutcome::RetryLocal),
             }
         }
+        let expected_incarnation = match odm_read_generation(req, bucket) {
+            Ok(Some(incarnation)) => incarnation,
+            Ok(None) => return None,
+            Err(err) => return Some(OdmGetOutcome::Respond(Err(err))),
+        };
+        match store.bucket_incarnation_id(bucket).await {
+            Ok(current) if current == expected_incarnation => {}
+            Ok(_) => return None,
+            Err(err) => return Some(OdmGetOutcome::Respond(Err(ApiError::from(err).into()))),
+        }
+        if !sys.is_module_enabled() {
+            return None;
+        }
+        let lookup = state.filter_incarnation(expected_incarnation)?.resolve_key(key)?;
+        let (state, client) = match odm_get_verdict(lookup) {
+            OdmGetVerdict::Fail(err) => return Some(OdmGetOutcome::Respond(Err(err))),
+            OdmGetVerdict::Consult { state, client } => (state, client),
+        };
         let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
         let reply = odm_get_from_source(&state, client.as_ref(), &req.headers, key, range, request_context).await;
         Some(match reply {
@@ -3895,7 +3913,7 @@ impl DefaultObjectUsecase {
         result
     }
 
-    async fn execute_get_object_inner(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
+    async fn execute_get_object_inner(&self, mut req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
         let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event();
 
         if let Some(context) = &self.context {
@@ -3981,6 +3999,8 @@ impl DefaultObjectUsecase {
                 return Self::complete_get_object_error(helper, err);
             }
         };
+        let bucket = req.input.bucket.clone();
+        prepare_odm_read_generation(&store, &mut req, &bucket).await;
         if let Some(request_context_start) = request_context_start {
             rustfs_io_metrics::record_get_object_stage_duration(
                 "s3_handler",
@@ -4919,7 +4939,7 @@ mod on_demand_migration_tests {
             Err(WriteBackError::Local("multipart is not part of the inline path".to_string()))
         }
 
-        async fn abort_multipart_upload(&self, _bucket: &str, _key: &str, _upload_id: &str) -> Result<(), WriteBackError> {
+        async fn abort_multipart_upload(&self, _request: &WriteBackRequest, _upload_id: &str) -> Result<(), WriteBackError> {
             Ok(())
         }
     }

rustfs/src/app/object/head.rs

@@ -146,6 +146,8 @@ impl DefaultObjectUsecase {
     /// `None` means the runtime does not intervene and the caller keeps its
     /// original 404. The source answer is never written back or queued.
     async fn on_demand_migration_head(
+        req: &S3Request<HeadObjectInput>,
+        store: &ECStore,
         bucket: &str,
         key: &str,
         opts: &ObjectOptions,
@@ -154,7 +156,33 @@ impl DefaultObjectUsecase {
         if !odm_request_may_consult_source(opts) {
             return None;
         }
-        let lookup = OnDemandMigrationSys::get().resolve(bucket, key)?;
+        let sys = OnDemandMigrationSys::get();
+        if !sys.is_module_enabled() {
+            return None;
+        }
+        let state = sys.state(bucket).filter(|state| state.matches_prefix(key))?;
+        let policy = &state.config().policy;
+        if !odm_policy_admits_miss(policy, miss) {
+            return None;
+        }
+        if policy.head == HeadPolicy::LocalOnly {
+            state.stats().record_request(OdmOp::Head, OdmOutcome::Filtered);
+            return None;
+        }
+        let expected_incarnation = match odm_read_generation(req, bucket) {
+            Ok(Some(incarnation)) => incarnation,
+            Ok(None) => return None,
+            Err(err) => return Some(Err(err)),
+        };
+        match store.bucket_incarnation_id(bucket).await {
+            Ok(current) if current == expected_incarnation => {}
+            Ok(_) => return None,
+            Err(err) => return Some(Err(ApiError::from(err).into())),
+        }
+        if !sys.is_module_enabled() {
+            return None;
+        }
+        let lookup = state.filter_incarnation(expected_incarnation)?.resolve_key(key)?;
         match odm_head_verdict(lookup, miss) {
             OdmHeadVerdict::Ignore => None,
             OdmHeadVerdict::Fail(err) => Some(Err(err)),
@@ -269,7 +297,7 @@ impl DefaultObjectUsecase {
     }
 
     #[instrument(level = "debug", skip(self, req))]
-    pub async fn execute_head_object(&self, req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
+    pub async fn execute_head_object(&self, mut req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
         if let Some(context) = &self.context {
             let _ = context.object_store();
         }
@@ -314,6 +342,8 @@ impl DefaultObjectUsecase {
             .await
             .map_err(ApiError::from)?;
 
+        prepare_odm_read_generation(&store, &mut req, &bucket).await;
+
         // Modification Points: Explicitly handles get_object_info errors, distinguishing between object absence and other errors
         let lookup = store.get_object_info(&bucket, &key, &opts).await;
         // Single classification point for the on-demand migration gate
@@ -347,7 +377,7 @@ impl DefaultObjectUsecase {
                         return result;
                     }
                     if let Some(miss) = odm_miss
-                        && let Some(result) = Self::on_demand_migration_head(&bucket, &key, &opts, miss).await
+                        && let Some(result) = Self::on_demand_migration_head(&req, &store, &bucket, &key, &opts, miss).await
                     {
                         return Self::finish_on_demand_migration_head(&req, &bucket, helper, result?).await;
                     }
@@ -362,7 +392,7 @@ impl DefaultObjectUsecase {
                 // A latest delete marker is a local miss the source may still
                 // answer when the bucket policy says so.
                 if let Some(miss) = odm_miss
-                    && let Some(result) = Self::on_demand_migration_head(&bucket, &key, &opts, miss).await
+                    && let Some(result) = Self::on_demand_migration_head(&req, &store, &bucket, &key, &opts, miss).await
                 {
                     return Self::finish_on_demand_migration_head(&req, &bucket, helper, result?).await;
                 }
@@ -1053,7 +1083,12 @@ mod tests {
             head: HeadPolicy::LocalOnly,
             ..Default::default()
         });
-        sys.apply(&bucket, Some(&cfg)).await;
+        sys.apply_for_incarnation(
+            &bucket,
+            store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+            Some(&cfg),
+        )
+        .await;
         let state = sys.state(&bucket).expect("bucket runtime installed");
 
         // Local hit: served locally, the runtime is never entered.
@@ -1109,7 +1144,12 @@ mod tests {
         );
 
         cfg.policy.respect_local_delete_marker = false;
-        sys.apply(&bucket, Some(&cfg)).await;
+        sys.apply_for_incarnation(
+            &bucket,
+            store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+            Some(&cfg),
+        )
+        .await;
         let err = Box::pin(usecase.execute_head_object(head_input(&bucket, "present", None)))
             .await
             .expect_err("local_only still answers 404");

rustfs/src/app/object/internal_put.rs

@@ -48,6 +48,8 @@ use http::HeaderName;
 /// other internal provenance is written verbatim.
 pub(crate) struct InternalPutContext {
     pub(crate) bucket: String,
+    /// Pins background work to its original bucket across deletion and recreation.
+    pub(crate) expected_bucket_incarnation_id: Option<Uuid>,
     pub(crate) key: String,
     /// Plaintext object length. The single-object path requires it, exactly
     /// like S3 PutObject rejects an unknown `Content-Length`.
@@ -239,6 +241,7 @@ impl DefaultObjectUsecase {
         let start_time = Instant::now();
         let InternalPutContext {
             bucket,
+            expected_bucket_incarnation_id,
             key,
             size,
             expected_md5_hex,
@@ -296,6 +299,7 @@ impl DefaultObjectUsecase {
                 principal_id,
                 emit_events,
                 preserve_delete_marker,
+                expected_bucket_incarnation_id,
             },
         };
         let committed = self
@@ -367,6 +371,8 @@ impl DefaultObjectUsecase {
             .await
             .map_err(ApiError::from)?;
 
+        opts.expected_bucket_incarnation_id = ctx.expected_bucket_incarnation_id;
+
         let dsc = must_replicate_object(
             &ctx.bucket,
             &ctx.key,
@@ -428,7 +434,10 @@ impl DefaultObjectUsecase {
         let bucket = ctx.bucket.as_str();
         let key = ctx.key.as_str();
         let store = self.object_store().ok_or_else(not_initialized)?;
-        let mut opts = ObjectOptions::default();
+        let mut opts = ObjectOptions {
+            expected_bucket_incarnation_id: ctx.expected_bucket_incarnation_id,
+            ..Default::default()
+        };
         let session = store
             .get_multipart_info(bucket, key, upload_id, &opts)
             .await
@@ -542,6 +551,7 @@ impl DefaultObjectUsecase {
         }
         let mut opts =
             get_complete_multipart_upload_opts_with_replication_authorization(&headers, false).map_err(ApiError::from)?;
+        opts.expected_bucket_incarnation_id = ctx.expected_bucket_incarnation_id;
         opts.preserve_etag = ctx.preserve_etag.clone();
         opts.preserve_delete_marker = ctx.preserve_delete_marker;
         let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
@@ -699,10 +709,24 @@ impl DefaultObjectUsecase {
     }
 
     /// Discard an internal multipart upload and its staged parts.
-    pub(crate) async fn internal_abort_multipart_upload(&self, bucket: &str, key: &str, upload_id: &str) -> Result<(), ApiError> {
+    pub(crate) async fn internal_abort_multipart_upload(
+        &self,
+        bucket: &str,
+        key: &str,
+        upload_id: &str,
+        expected_bucket_incarnation_id: Option<Uuid>,
+    ) -> Result<(), ApiError> {
         let store = self.object_store().ok_or_else(not_initialized)?;
         store
-            .abort_multipart_upload(bucket, key, upload_id, &ObjectOptions::default())
+            .abort_multipart_upload(
+                bucket,
+                key,
+                upload_id,
+                &ObjectOptions {
+                    expected_bucket_incarnation_id,
+                    ..Default::default()
+                },
+            )
             .await
             .map_err(ApiError::from)?;
         rustfs_scanner::record_dirty_usage_bucket(bucket);
@@ -756,6 +780,7 @@ mod tests {
     fn internal_context(bucket: &str, key: &str, body: &[u8]) -> InternalPutContext {
         InternalPutContext {
             bucket: bucket.to_string(),
+            expected_bucket_incarnation_id: None,
             key: key.to_string(),
             size: Some(body.len() as u64),
             expected_md5_hex: Some(md5_hex(body)),
@@ -1138,9 +1163,14 @@ mod tests {
         ))
         .await
         .expect("part of the aborted upload must stage");
-        Box::pin(usecase.internal_abort_multipart_upload(&bucket, &ctx.key, &aborted_upload_id))
+        Box::pin(usecase.internal_abort_multipart_upload(
-            .await
+            &bucket,
-            .expect("internal abort must succeed");
+            &ctx.key,
+            &aborted_upload_id,
+            ctx.expected_bucket_incarnation_id,
+        ))
+        .await
+        .expect("internal abort must succeed");
         let uploads = Box::pin(store.list_multipart_uploads(&bucket, &ctx.key, None, None, None, 100))
             .await
             .expect("list multipart uploads after abort");

rustfs/src/app/object/mod.rs

@@ -21,8 +21,8 @@ use crate::storage_api::table::get_bucket_metadata;
 
 use super::storage_api::object_usecase::access::{
     PostObjectRequestMarker, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard, authorize_request,
-    has_bypass_governance_header, load_bucket_generation_from_store, recursive_force_delete_is_authorized,
+    has_bypass_governance_header, load_bucket_generation_from_store, odm_read_generation, prepare_odm_read_generation,
-    replication_request_authorized, req_info_mut, req_info_ref,
+    recursive_force_delete_is_authorized, replication_request_authorized, req_info_mut, req_info_ref,
 };
 #[cfg(test)]
 use super::storage_api::object_usecase::bucket::quota::BucketQuota;

rustfs/src/app/object/on_demand_migration_put.rs

@@ -172,6 +172,7 @@ pub(super) async fn write_back_context(request: &WriteBackRequest, single_part:
     };
     InternalPutContext {
         bucket: request.bucket.clone(),
+        expected_bucket_incarnation_id: Some(request.bucket_incarnation_id),
         key: request.key.clone(),
         size: Some(head.size),
         expected_md5_hex: single_part.then(|| expected_md5_hex(head)).flatten(),
@@ -285,9 +286,9 @@ impl OdmWriteBack for OnDemandMigrationWriteBack {
             .map_err(write_back_error)
     }
 
-    async fn abort_multipart_upload(&self, bucket: &str, key: &str, upload_id: &str) -> Result<(), WriteBackError> {
+    async fn abort_multipart_upload(&self, request: &WriteBackRequest, upload_id: &str) -> Result<(), WriteBackError> {
         self.usecase()
-            .internal_abort_multipart_upload(bucket, key, upload_id)
+            .internal_abort_multipart_upload(&request.bucket, &request.key, upload_id, Some(request.bucket_incarnation_id))
             .await
             .map_err(write_back_error)
     }
@@ -303,7 +304,7 @@ mod tests {
         ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration,
     };
     use crate::app::storage_api::test::bucket::utils::serialize;
-    use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+    use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
     use crate::app::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata};
     use crate::on_demand_migration::{PullFailureReason, SourceSse};
     use http::Method;
@@ -345,9 +346,10 @@ mod tests {
         }
     }
 
-    fn request(bucket: &str, key: &str, head: SourceHead) -> WriteBackRequest {
+    fn request(bucket: &str, bucket_incarnation_id: Uuid, key: &str, head: SourceHead) -> WriteBackRequest {
         WriteBackRequest {
             bucket: bucket.to_string(),
+            bucket_incarnation_id,
             key: key.to_string(),
             head,
             source_label: SOURCE_LABEL.to_string(),
@@ -475,7 +477,15 @@ mod tests {
         let body = b"pulled from the legacy bucket".to_vec();
         let head = source_head(&body);
         let outcome = write_back
-            .put_object(&request(&bucket, "dir/obj.txt", head.clone()), body_stream(&body))
+            .put_object(
+                &request(
+                    &bucket,
+                    store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                    "dir/obj.txt",
+                    head.clone(),
+                ),
+                body_stream(&body),
+            )
             .await
             .expect("write-back must commit");
         assert_eq!(outcome.etag, head.etag, "single-part source ETag is preserved");
@@ -518,7 +528,12 @@ mod tests {
             .await
             .expect("bucket");
         let write_back = OnDemandMigrationWriteBack::new();
-        let req = request(bucket, "key", source_head(b"source"));
+        let req = request(
+            bucket,
+            store.bucket_incarnation_id(bucket).await.expect("bucket incarnation"),
+            "key",
+            source_head(b"source"),
+        );
         assert!(matches!(
             write_back.put_object(&req, body_stream(b"source")).await,
             Err(WriteBackError::Unsupported(_))
@@ -534,6 +549,75 @@ mod tests {
         assert_nothing_left(&store, bucket, "key").await;
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn stale_write_back_cannot_mutate_a_recreated_bucket() {
+        let (store, bucket) = write_back_test_bucket("odm-wb-incarnation", false).await;
+        let old_id = store.bucket_incarnation_id(&bucket).await.expect("old bucket incarnation");
+        let stale = request(&bucket, old_id, "object", source_head(b"source"));
+        let (resume, wait) = tokio::sync::oneshot::channel();
+        let delayed = {
+            let stale = stale.clone();
+            tokio::spawn(async move {
+                wait.await.expect("resume old source pull");
+                OnDemandMigrationWriteBack::new()
+                    .put_object(&stale, body_stream(b"source"))
+                    .await
+            })
+        };
+        store
+            .delete_bucket(
+                &bucket,
+                &DeleteBucketOptions {
+                    force: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("delete original bucket");
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("recreate bucket");
+        let new_id = store.bucket_incarnation_id(&bucket).await.expect("replacement incarnation");
+        assert_ne!(old_id, new_id);
+        resume.send(()).expect("release delayed pull");
+        assert!(delayed.await.expect("delayed pull task").is_err());
+        assert_nothing_left(&store, &bucket, "object").await;
+
+        let write_back = OnDemandMigrationWriteBack::new();
+        assert!(write_back.create_multipart_upload(&stale).await.is_err());
+        let current = request(&bucket, new_id, "object", source_head(b"current"));
+        let upload = write_back
+            .create_multipart_upload(&current)
+            .await
+            .expect("create replacement upload");
+        // A stale capability must fail independently of whether its upload ID
+        // happens to name a valid session in the replacement bucket.
+        assert!(
+            write_back
+                .upload_part(&stale, &upload, 1, 6, body_stream(b"source"))
+                .await
+                .is_err()
+        );
+        let part = write_back
+            .upload_part(&current, &upload, 1, 7, body_stream(b"current"))
+            .await
+            .expect("stage current part");
+        assert!(
+            write_back
+                .complete_multipart_upload(&stale, &upload, vec![part.clone()])
+                .await
+                .is_err()
+        );
+        assert!(write_back.abort_multipart_upload(&stale, &upload).await.is_err());
+        write_back
+            .complete_multipart_upload(&current, &upload, vec![part])
+            .await
+            .expect("stale cleanup preserves replacement upload");
+        assert_eq!(raw_object_bytes(&store, &bucket, "object").await, b"current");
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn write_back_integrity_failure_leaves_nothing_behind() {
@@ -543,7 +627,15 @@ mod tests {
         head.etag = Some(md5_hex(b"a different body"));
 
         let err = OnDemandMigrationWriteBack::new()
-            .put_object(&request(&bucket, "wrong.bin", head), body_stream(&body))
+            .put_object(
+                &request(
+                    &bucket,
+                    store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                    "wrong.bin",
+                    head,
+                ),
+                body_stream(&body),
+            )
             .await
             .expect_err("an ETag mismatch must fail the write-back");
         assert_eq!(err, WriteBackError::Integrity);
@@ -559,8 +651,18 @@ mod tests {
             let (store, bucket) = write_back_test_bucket("odm-wb-race", versioned).await;
             let source = b"old source bytes";
             let client = b"new client bytes";
-            let req = request(&bucket, "race", source_head(source));
+            let req = request(
-            let client_req = request(&bucket, "race", source_head(client));
+                &bucket,
+                store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                "race",
+                source_head(source),
+            );
+            let client_req = request(
+                &bucket,
+                store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                "race",
+                source_head(client),
+            );
             let mut client_ctx = write_back_context(&client_req, true).await;
             client_ctx.if_absent = false;
             let client_after = PutObjectCommitBarrier::install(&bucket, "race", PutObjectCommitPause::AfterNamespace);
@@ -595,13 +697,27 @@ mod tests {
     async fn write_back_multipart_completion_preserves_a_client_put_after_staging() {
         let (store, bucket) = write_back_test_bucket("odm-mpu-race", false).await;
         let write_back = OnDemandMigrationWriteBack::new();
-        let req = request(&bucket, "race", source_head(b"source"));
+        let req = request(
+            &bucket,
+            store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+            "race",
+            source_head(b"source"),
+        );
         let upload_id = write_back.create_multipart_upload(&req).await.expect("create");
         let part = write_back
             .upload_part(&req, &upload_id, 1, 6, body_stream(b"source"))
             .await
             .expect("stage");
-        let mut client_ctx = write_back_context(&request(&bucket, "race", source_head(b"client")), true).await;
+        let mut client_ctx = write_back_context(
+            &request(
+                &bucket,
+                store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                "race",
+                source_head(b"client"),
+            ),
+            true,
+        )
+        .await;
         client_ctx.if_absent = false;
         let committed = DefaultObjectUsecase::from_global()
             .internal_put_object(client_ctx, body_stream(b"client"))
@@ -613,7 +729,7 @@ mod tests {
             "{result:?}"
         );
         write_back
-            .abort_multipart_upload(&bucket, "race", &upload_id)
+            .abort_multipart_upload(&req, &upload_id)
             .await
             .expect("abort rejected upload");
         let stored = stored_object(&store, &bucket, "race").await;
@@ -628,7 +744,12 @@ mod tests {
         for multipart in [false, true] {
             let (store, bucket) = write_back_test_bucket("odm-wb-tombstone", true).await;
             let write_back = OnDemandMigrationWriteBack::new();
-            let mut req = request(&bucket, "deleted", source_head(b"source"));
+            let mut req = request(
+                &bucket,
+                store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                "deleted",
+                source_head(b"source"),
+            );
             let staged = if multipart {
                 let id = write_back.create_multipart_upload(&req).await.expect("create");
                 let part = write_back
@@ -654,10 +775,7 @@ mod tests {
             assert!(marker.delete_marker);
             let rejected = if let Some((id, part)) = staged {
                 let result = write_back.complete_multipart_upload(&req, &id, vec![part]).await;
-                write_back
+                write_back.abort_multipart_upload(&req, &id).await.expect("abort");
-                    .abort_multipart_upload(&bucket, "deleted", &id)
-                    .await
-                    .expect("abort");
                 result
             } else {
                 write_back.put_object(&req, body_stream(b"source")).await
@@ -691,7 +809,15 @@ mod tests {
             Err(io::Error::new(io::ErrorKind::BrokenPipe, "tee primary dropped before EOF")),
         ]);
         let err = OnDemandMigrationWriteBack::new()
-            .put_object(&request(&bucket, "torn.bin", head.clone()), torn)
+            .put_object(
+                &request(
+                    &bucket,
+                    store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                    "torn.bin",
+                    head.clone(),
+                ),
+                torn,
+            )
             .await
             .expect_err("a broken stream must fail the write-back");
         assert_ne!(err, WriteBackError::Integrity, "{err}");
@@ -700,7 +826,15 @@ mod tests {
         // A clean EOF short of the advertised size is just as fatal.
         let short = stream(vec![Ok(Bytes::copy_from_slice(&body[..64 * 1024]))]);
         let err = OnDemandMigrationWriteBack::new()
-            .put_object(&request(&bucket, "short.bin", head), short)
+            .put_object(
+                &request(
+                    &bucket,
+                    store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                    "short.bin",
+                    head,
+                ),
+                short,
+            )
             .await
             .expect_err("a short body must fail the write-back");
         assert!(matches!(err, WriteBackError::Local(_) | WriteBackError::Integrity), "{err}");
@@ -716,7 +850,12 @@ mod tests {
         let mut head = source_head(&body);
         head.etag = Some(format!("{}-2", md5_hex(&body)));
         head.is_multipart_etag = true;
-        let request = request(&bucket, "big/object.bin", head.clone());
+        let request = request(
+            &bucket,
+            store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+            "big/object.bin",
+            head.clone(),
+        );
         let write_back = OnDemandMigrationWriteBack::new();
 
         let upload_id = write_back.create_multipart_upload(&request).await.expect("create");
@@ -754,10 +893,7 @@ mod tests {
             .upload_part(&request, &aborted, 1, 4096, body_stream(&body[..4096]))
             .await
             .expect("stage part");
-        write_back
+        write_back.abort_multipart_upload(&request, &aborted).await.expect("abort");
-            .abort_multipart_upload(&bucket, "big/object.bin", &aborted)
-            .await
-            .expect("abort");
         let uploads = store
             .list_multipart_uploads(&bucket, "big/object.bin", None, None, None, 100)
             .await
@@ -810,7 +946,12 @@ mod tests {
 
             let body = b"plaintext that must be encrypted at rest".to_vec();
             let head = source_head(&body);
-            let request = request(&bucket, "secret.txt", head.clone());
+            let request = request(
+                &bucket,
+                store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                "secret.txt",
+                head.clone(),
+            );
             // The source ETag is not forced onto an encrypted object; the
             // local ETag is whatever the SSE write path computes.
             assert_eq!(write_back_context(&request, true).await.preserve_etag, None);
@@ -848,7 +989,15 @@ mod tests {
         let (store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket("odm-wb-quota", 64).await;
         let body = vec![0x71; 4096];
         let err = OnDemandMigrationWriteBack::new()
-            .put_object(&request(&bucket, "over.bin", source_head(&body)), body_stream(&body))
+            .put_object(
+                &request(
+                    &bucket,
+                    store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+                    "over.bin",
+                    source_head(&body),
+                ),
+                body_stream(&body),
+            )
             .await
             .expect_err("a full quota must reject the write-back");
         assert!(matches!(err, WriteBackError::Quota(_)), "{err}");
@@ -913,7 +1062,12 @@ mod tests {
 
         let body = b"replicate me".to_vec();
         let head = source_head(&body);
-        let request = request(&bucket, "replicated.txt", head);
+        let request = request(
+            &bucket,
+            store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+            "replicated.txt",
+            head,
+        );
         let ctx = write_back_context(&request, true).await;
         assert!(ctx.emit_events, "policy.emit_events reaches the creation event");
         assert_eq!(ctx.principal_id, ON_DEMAND_MIGRATION_PRINCIPAL_ID);
@@ -961,7 +1115,12 @@ mod tests {
         head.user_metadata
             .insert(forged_replica_key.to_string(), ReplicationStatusType::Replica.as_str().to_string());
 
-        let mut write_request = request(&bucket, "unadmitted.txt", head);
+        let mut write_request = request(
+            &bucket,
+            store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"),
+            "unadmitted.txt",
+            head,
+        );
         write_request.tags = Some(HashMap::from([("replicate".to_string(), "no".to_string())]));
         OnDemandMigrationWriteBack::new()
             .put_object(&write_request, body_stream(&body))
@@ -1059,7 +1218,7 @@ mod tests {
 
     #[test]
     fn provenance_and_tags_are_stable() {
-        let mut request = request("b", "k", source_head(b"x"));
+        let mut request = request("b", Uuid::nil(), "k", source_head(b"x"));
         let metadata = provenance_metadata(&request);
         assert_eq!(metadata.len(), 10, "five keys under two prefixes");
         assert_provenance(&metadata, &request.head);
@@ -1080,7 +1239,7 @@ mod tests {
     #[tokio::test]
     async fn write_back_context_applies_the_etag_and_event_policy() {
         let body = b"context".to_vec();
-        let mut request = request("no-such-bucket", "k", source_head(&body));
+        let mut request = request("no-such-bucket", Uuid::nil(), "k", source_head(&body));
         let ctx = write_back_context(&request, true).await;
         assert_eq!(ctx.expected_md5_hex, Some(md5_hex(&body)));
         assert_eq!(ctx.preserve_etag, Some(md5_hex(&body)));

rustfs/src/app/object/put.rs

@@ -953,6 +953,7 @@ pub(super) enum PutObjectOrigin<'a> {
         principal_id: &'static str,
         emit_events: bool,
         preserve_delete_marker: bool,
+        expected_bucket_incarnation_id: Option<Uuid>,
     },
 }
 
@@ -967,7 +968,13 @@ impl PutObjectOrigin<'_> {
     fn apply_bucket_generation_guard(&self, bucket: &str, opts: &mut ObjectOptions) -> S3Result<()> {
         match self {
             Self::S3 { req, .. } => apply_bucket_generation_guard(req, bucket, opts),
-            Self::Internal { .. } => Ok(()),
+            Self::Internal {
+                expected_bucket_incarnation_id,
+                ..
+            } => {
+                opts.expected_bucket_incarnation_id = *expected_bucket_incarnation_id;
+                Ok(())
+            }
         }
     }
 

rustfs/src/app/storage_api.rs

@@ -27,19 +27,20 @@ pub(crate) fn EndpointServerPools(
 /// S3 wire types for app-layer modules, funneled here so new files stay off
 /// the direct s3s surface (s3s footprint ratchet, `scripts/check_s3s_footprint.sh`).
 pub(crate) mod s3 {
+    #[cfg(test)]
+    pub(crate) use s3s::S3Response;
     #[cfg(test)]
     pub(crate) use s3s::dto::ListObjectsInput;
     #[cfg(test)]
     pub(crate) use s3s::dto::{
-        BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ListObjectsV2Input,
+        BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, GetObjectInput,
-        ListObjectsV2Output, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus,
+        HeadObjectInput, ListObjectsV2Input, ListObjectsV2Output, ReplicationConfiguration, ReplicationRule,
-        ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration,
+        ReplicationRuleFilter, ReplicationRuleStatus, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration,
+        ServerSideEncryptionRule, Tag, VersioningConfiguration,
     };
     #[cfg(test)]
     pub(crate) use s3s::xml::{Serialize as XmlSerialize, Serializer as XmlSerializer};
-    pub(crate) use s3s::{S3Error, S3ErrorCode, S3Result};
+    pub(crate) use s3s::{S3Error, S3ErrorCode, S3Request, S3Result};
-    #[cfg(test)]
-    pub(crate) use s3s::{S3Request, S3Response};
 }
 
 pub(crate) mod admin {
@@ -266,8 +267,9 @@ pub(crate) mod access {
     pub(crate) use crate::storage::storage_api::access_consumer::{
         PostObjectRequestMarker, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard, authorize_request,
         bucket_config_mutation_incarnation, has_bypass_governance_header, load_bucket_generation_from_store,
-        log_list_buckets_iam_implicit_deny, prepare_list_buckets_iam_authorization, recursive_force_delete_is_authorized,
+        log_list_buckets_iam_implicit_deny, odm_read_generation, prepare_list_buckets_iam_authorization,
-        replication_request_authorized, req_info_mut, req_info_ref,
+        prepare_odm_read_generation, recursive_force_delete_is_authorized, replication_request_authorized, req_info_mut,
+        req_info_ref,
     };
 }
 

rustfs/src/on_demand_migration/backfill.rs

@@ -370,6 +370,8 @@ pub type PullReport = Option<super::pull::QueuedPullReport>;
 /// mock in unit tests. Production: [`BucketBackfillContext`].
 #[async_trait]
 pub trait BackfillContext: Send + Sync {
+    /// The bucket incarnation captured by this context.
+    fn incarnation_id(&self) -> Uuid;
     /// One source page in the local key namespace.
     async fn list_page(&self, prefix: Option<&str>, token: Option<&str>, max_keys: i32) -> Result<SourcePage, SourceError>;
     /// Whether the breaker admits source traffic right now.
@@ -411,6 +413,10 @@ impl BucketBackfillContext {
 
 #[async_trait]
 impl BackfillContext for BucketBackfillContext {
+    fn incarnation_id(&self) -> Uuid {
+        self.state.incarnation_id()
+    }
+
     async fn list_page(&self, prefix: Option<&str>, token: Option<&str>, max_keys: i32) -> Result<SourcePage, SourceError> {
         let client = self.state.client().map_err(|err| SourceError::Unsupported(err.to_string()))?;
         let started = Instant::now();
@@ -661,8 +667,36 @@ pub async fn read_checkpoint(api: &Arc<ECStore>, bucket: &str) -> Result<Option<
 async fn write_checkpoint(
     api: &Arc<ECStore>,
     bucket: &str,
+    incarnation_id: Uuid,
     checkpoint: &BackfillCheckpoint,
     expected_etag: Option<&str>,
+) -> Result<String, BackfillError> {
+    let api = Arc::clone(api);
+    let bucket = bucket.to_string();
+    let checkpoint = checkpoint.clone();
+    let expected_etag = expected_etag.map(str::to_string);
+    // The storage commit owns detached work. Keep its user-bucket fence alive
+    // even when a caller aborts its waiter before the erasure tail has drained.
+    tokio::spawn(async move {
+        let fence = api.acquire_bucket_incarnation_fence(&bucket, incarnation_id).await?;
+        let mut opts = ObjectOptions::default();
+        fence.attach_to_object_options(&mut opts);
+        let result = write_checkpoint_while_fenced(&api, &bucket, &checkpoint, expected_etag.as_deref(), opts).await;
+        drop(fence);
+        result
+    })
+    .await
+    .map_err(|err| StorageError::other(format!("backfill checkpoint task failed: {err}")))?
+}
+
+/// The caller holds the destination bucket's lifecycle fence through the CAS
+/// write and its read-back, including the drained erasure write tail.
+async fn write_checkpoint_while_fenced(
+    api: &Arc<ECStore>,
+    bucket: &str,
+    checkpoint: &BackfillCheckpoint,
+    expected_etag: Option<&str>,
+    mut opts: ObjectOptions,
 ) -> Result<String, BackfillError> {
     let data = checkpoint.to_json()?;
     let preconditions = match expected_etag {
@@ -675,12 +709,9 @@ async fn write_checkpoint(
             ..Default::default()
         },
     };
-    let opts = ObjectOptions {
+    opts.max_parity = true;
-        max_parity: true,
+    opts.write_completion = WriteCompletion::TailDrained;
-        write_completion: WriteCompletion::TailDrained,
+    opts.http_preconditions = Some(preconditions);
-        http_preconditions: Some(preconditions),
-        ..Default::default()
-    };
     match save_config_with_opts(Arc::clone(api), &checkpoint_path(bucket), data, &opts).await {
         Ok(()) => {}
         Err(StorageError::PreconditionFailed) => return Err(BackfillError::Conflict(bucket.to_string())),
@@ -851,7 +882,14 @@ impl BackfillRunner {
             });
         }
         let checkpoint = BackfillCheckpoint::new(&request, config_updated_at, &self.node, now);
-        let etag = write_checkpoint(&self.api, bucket, &checkpoint, stored.as_ref().map(|s| s.etag.as_str())).await?;
+        let etag = write_checkpoint(
+            &self.api,
+            bucket,
+            context.incarnation_id(),
+            &checkpoint,
+            stored.as_ref().map(|s| s.etag.as_str()),
+        )
+        .await?;
         info!(
             event = EVENT_ODM_BACKFILL_STATE,
             component = LOG_COMPONENT_ECSTORE,
@@ -886,30 +924,42 @@ impl BackfillRunner {
             }
             return Ok(handle.snapshot.lock().clone());
         }
-        let _lock = self.lease_lock(bucket, get_lock_acquire_timeout()).await?;
+        let incarnation_id = self.api.bucket_incarnation_id_from_disk(bucket).await?;
-        let Some(stored) = read_checkpoint(&self.api, bucket).await? else {
+        let lock = self.lease_lock(bucket, get_lock_acquire_timeout()).await?;
-            return Err(BackfillError::NotFound(bucket.to_string()));
+        let api = Arc::clone(&self.api);
-        };
+        let bucket = bucket.to_string();
-        if !stored.checkpoint.state.is_active() {
+        tokio::spawn(async move {
-            return Ok(stored.checkpoint);
+            let _lock = lock;
-        }
+            let fence = api.acquire_bucket_incarnation_fence(&bucket, incarnation_id).await?;
-        let mut checkpoint = stored.checkpoint;
+            let mut opts = ObjectOptions::default();
-        let now = OffsetDateTime::now_utc();
+            fence.attach_to_object_options(&mut opts);
-        checkpoint.state = BackfillState::Cancelled;
+            let Some(stored) = read_checkpoint(&api, &bucket).await? else {
-        checkpoint.updated_at = now;
+                return Err(BackfillError::NotFound(bucket.to_string()));
-        write_checkpoint(&self.api, bucket, &checkpoint, Some(&stored.etag)).await?;
+            };
-        info!(
+            if !stored.checkpoint.state.is_active() {
-            event = EVENT_ODM_BACKFILL_STATE,
+                return Ok(stored.checkpoint);
-            component = LOG_COMPONENT_ECSTORE,
+            }
-            subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION,
+            let mut checkpoint = stored.checkpoint;
-            state = checkpoint.state.as_str(),
+            let now = OffsetDateTime::now_utc();
-            result = "cancelled",
+            checkpoint.state = BackfillState::Cancelled;
-            bucket = %bucket,
+            checkpoint.updated_at = now;
-            job_id = %checkpoint.job_id,
+            write_checkpoint_while_fenced(&api, &bucket, &checkpoint, Some(&stored.etag), opts).await?;
-            owner = %checkpoint.owner.as_ref().map(|o| o.node.as_str()).unwrap_or_default(),
+            info!(
-            "On-demand migration backfill job cancelled remotely"
+                event = EVENT_ODM_BACKFILL_STATE,
-        );
+                component = LOG_COMPONENT_ECSTORE,
-        Ok(checkpoint)
+                subsystem = LOG_SUBSYSTEM_ON_DEMAND_MIGRATION,
+                state = checkpoint.state.as_str(),
+                result = "cancelled",
+                bucket = %bucket,
+                job_id = %checkpoint.job_id,
+                owner = %checkpoint.owner.as_ref().map(|o| o.node.as_str()).unwrap_or_default(),
+                "On-demand migration backfill job cancelled remotely"
+            );
+            drop(fence);
+            Ok(checkpoint)
+        })
+        .await
+        .map_err(|err| StorageError::other(format!("backfill cancellation task failed: {err}")))?
     }
 
     /// Latest checkpoint: the in-memory progress of a local job, else the
@@ -1002,7 +1052,7 @@ impl BackfillRunner {
             checkpoint.state = BackfillState::Cancelled;
             checkpoint.updated_at = now;
             checkpoint.record_failure("config_changed", None, now);
-            write_checkpoint(&self.api, bucket, &checkpoint, Some(&stored.etag)).await?;
+            write_checkpoint(&self.api, bucket, context.incarnation_id(), &checkpoint, Some(&stored.etag)).await?;
             info!(
                 event = EVENT_ODM_BACKFILL_STATE,
                 component = LOG_COMPONENT_ECSTORE,
@@ -1021,7 +1071,7 @@ impl BackfillRunner {
             node: self.node.clone(),
             lease_until: now + BACKFILL_LEASE,
         });
-        let etag = write_checkpoint(&self.api, bucket, &checkpoint, Some(&stored.etag)).await?;
+        let etag = write_checkpoint(&self.api, bucket, context.incarnation_id(), &checkpoint, Some(&stored.etag)).await?;
         warn!(
             event = EVENT_ODM_BACKFILL_LEASE_TAKEOVER,
             component = LOG_COMPONENT_ECSTORE,
@@ -1460,7 +1510,8 @@ impl Job {
                 lease_until: now + BACKFILL_LEASE,
             });
         }
-        let etag = write_checkpoint(&self.api, &self.bucket, &self.checkpoint, Some(&self.etag)).await?;
+        let etag =
+            write_checkpoint(&self.api, &self.bucket, self.context.incarnation_id(), &self.checkpoint, Some(&self.etag)).await?;
         self.etag = etag;
         self.keys_since_save = 0;
         self.last_save = Instant::now();
@@ -1500,7 +1551,10 @@ pub async fn run_backfill_recovery_loop(runner: Arc<BackfillRunner>, cancel: Can
 
 #[cfg(test)]
 mod tests {
-    use super::super::storage_api::test_support::isolated_store_over_temp_disks;
+    use super::super::storage_api::test_support::{
+        BUCKET_LIFECYCLE_LOCK_OBJECT, BucketOperations as _, PutObjectCommitBarrier, PutObjectCommitPause,
+        isolated_store_over_temp_disks,
+    };
     use super::*;
     use crate::on_demand_migration::source_client::SourceObject;
     use crate::on_demand_migration::sys::PullError;
@@ -1626,6 +1680,7 @@ mod tests {
 
     /// Scripted source + local store + queue with a controllable report path.
     struct MockContext {
+        incarnation_id: Mutex<Option<Uuid>>,
         objects: Vec<SourceObject>,
         page_size: usize,
         local: Mutex<HashMap<String, LocalBackfillObject>>,
@@ -1654,6 +1709,7 @@ mod tests {
                 })
                 .collect();
             Arc::new(Self {
+                incarnation_id: Mutex::new(None),
                 objects,
                 page_size,
                 local: Mutex::new(HashMap::new()),
@@ -1688,6 +1744,10 @@ mod tests {
 
     #[async_trait]
     impl BackfillContext for MockContext {
+        fn incarnation_id(&self) -> Uuid {
+            self.incarnation_id.lock().expect("test bucket initialized")
+        }
+
         async fn list_page(&self, prefix: Option<&str>, token: Option<&str>, max_keys: i32) -> Result<SourcePage, SourceError> {
             if let Some(err) = self.list_error.lock().take() {
                 return Err(err);
@@ -1780,12 +1840,17 @@ mod tests {
         context: Arc<MockContext>,
     ) -> (Vec<tempfile::TempDir>, Arc<ECStore>, Arc<BackfillRunner>) {
         let (dirs, store) = isolated_store_over_temp_disks().await;
-        // The isolated store has no bucket metadata system; the checkpoint
+        super::super::storage_api::test_support::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
-        // only needs the bucket's directory under the metadata volume.
+        store
-        for dir in &dirs {
+            .make_bucket(bucket, &Default::default())
-            std::fs::create_dir_all(dir.path().join(RUSTFS_META_BUCKET).join(BUCKET_META_PREFIX).join(bucket))
+            .await
-                .expect("test bucket metadata directory");
+            .expect("create test bucket");
-        }
+        *context.incarnation_id.lock() = Some(
+            store
+                .bucket_incarnation_id_from_disk(bucket)
+                .await
+                .expect("test bucket identity"),
+        );
         let runner = runner_on(node, bucket, context, Arc::clone(&store));
         (dirs, store, runner)
     }
@@ -1795,6 +1860,129 @@ mod tests {
         BackfillRunner::new(store, node, Arc::new(contexts))
     }
 
+    #[tokio::test]
+    async fn cancelled_checkpoint_waiter_keeps_bucket_fenced_until_commit_finishes() {
+        for (suffix, pause) in [
+            ("before", PutObjectCommitPause::BeforeQuotaRename),
+            ("after", PutObjectCommitPause::AfterRenameQuorum),
+        ] {
+            let bucket = format!("backfill-cancel-tail-{suffix}");
+            let context = MockContext::new(0, 1);
+            let (_dirs, store, _runner) = runner_with("node-a", &bucket, Arc::clone(&context)).await;
+            let original_incarnation = context.incarnation_id();
+            let checkpoint = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-a", ts(1_700_000_001));
+            let barrier = PutObjectCommitBarrier::install(RUSTFS_META_BUCKET, &checkpoint_path(&bucket), pause);
+            let writer_api = Arc::clone(&store);
+            let writer_bucket = bucket.clone();
+            let waiter = tokio::spawn(async move {
+                write_checkpoint(&writer_api, &writer_bucket, original_incarnation, &checkpoint, None).await
+            });
+            barrier.wait_until_paused().await;
+            waiter.abort();
+            assert!(waiter.await.expect_err("caller aborted").is_cancelled());
+
+            let lifecycle_lock = store
+                .new_ns_lock(&bucket, BUCKET_LIFECYCLE_LOCK_OBJECT)
+                .await
+                .expect("lifecycle lock");
+            {
+                let mut probe = Box::pin(lifecycle_lock.get_write_lock(Duration::from_secs(1)));
+                assert!(
+                    futures::poll!(probe.as_mut()).is_pending(),
+                    "lifecycle writer must first try to acquire the lock"
+                );
+                assert!(
+                    tokio::time::timeout(Duration::from_millis(100), probe.as_mut())
+                        .await
+                        .is_err(),
+                    "the checkpoint owner must retain the user bucket lifecycle read lock after caller cancellation"
+                );
+            }
+
+            let delete_api = Arc::clone(&store);
+            let delete_bucket = bucket.clone();
+            let mut deletion = tokio::spawn(async move { delete_api.delete_bucket(&delete_bucket, &Default::default()).await });
+            assert!(
+                tokio::time::timeout(Duration::from_millis(100), &mut deletion).await.is_err(),
+                "DeleteBucket must wait for the checkpoint owner after its caller aborts"
+            );
+            barrier.release();
+            tokio::time::timeout(Duration::from_secs(10), deletion)
+                .await
+                .expect("commit must drain and release its lifecycle guard")
+                .expect("delete task")
+                .expect("delete original bucket");
+            store
+                .make_bucket(&bucket, &Default::default())
+                .await
+                .expect("recreate bucket");
+            assert_ne!(
+                original_incarnation,
+                store.bucket_incarnation_id_from_disk(&bucket).await.expect("new identity")
+            );
+            assert!(
+                read_checkpoint(&store, &bucket)
+                    .await
+                    .expect("read recreated bucket")
+                    .is_none(),
+                "no old checkpoint may outlive bucket deletion"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn stale_checkpoint_writer_cannot_resurrect_or_overwrite_a_recreated_bucket() {
+        let bucket = "backfill-incarnation";
+        let context = MockContext::new(0, 1);
+        let (_dirs, store, _runner) = runner_with("node-a", bucket, Arc::clone(&context)).await;
+        let old_incarnation = context.incarnation_id();
+        let old = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-a", ts(1_700_000_001));
+        let old_etag = write_checkpoint(&store, bucket, old_incarnation, &old, None)
+            .await
+            .expect("old checkpoint");
+
+        store
+            .delete_bucket(bucket, &Default::default())
+            .await
+            .expect("delete original bucket");
+        store.make_bucket(bucket, &Default::default()).await.expect("recreate bucket");
+        let current_incarnation = store.bucket_incarnation_id_from_disk(bucket).await.expect("new identity");
+        assert_ne!(old_incarnation, current_incarnation);
+        assert!(
+            read_checkpoint(&store, bucket)
+                .await
+                .expect("read after recreation")
+                .is_none()
+        );
+        for expected_etag in [None, Some(old_etag.as_str())] {
+            let error = write_checkpoint(&store, bucket, old_incarnation, &old, expected_etag)
+                .await
+                .expect_err("stale writer rejected");
+            assert!(matches!(error, BackfillError::Storage(StorageError::BucketNotFound(_))));
+        }
+        assert!(
+            read_checkpoint(&store, bucket)
+                .await
+                .expect("stale writer left no checkpoint")
+                .is_none()
+        );
+
+        let current = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-b", ts(1_700_000_002));
+        let current_etag = write_checkpoint(&store, bucket, current_incarnation, &current, None)
+            .await
+            .expect("current checkpoint");
+        let error = write_checkpoint(&store, bucket, old_incarnation, &old, Some(&current_etag))
+            .await
+            .expect_err("old identity cannot overwrite a matching ETag");
+        assert!(matches!(error, BackfillError::Storage(StorageError::BucketNotFound(_))));
+        let stored = read_checkpoint(&store, bucket)
+            .await
+            .expect("read current checkpoint")
+            .expect("current checkpoint remains");
+        assert_eq!(stored.etag, current_etag);
+        assert_eq!(stored.checkpoint, current);
+    }
+
     #[tokio::test]
     async fn full_backfill_lists_pages_and_counts_every_key() {
         let bucket = "backfill-full";
@@ -2134,7 +2322,7 @@ mod tests {
             node: "node-a".to_string(),
             lease_until: now - Duration::from_secs(120),
         });
-        let etag = write_checkpoint(&store, bucket, &crashed, None)
+        let etag = write_checkpoint(&store, bucket, context.incarnation_id(), &crashed, None)
             .await
             .expect("seed checkpoint");
 
@@ -2145,7 +2333,7 @@ mod tests {
             lease_until: now + Duration::from_secs(60),
         });
         live.updated_at = now;
-        let etag = write_checkpoint(&store, bucket, &live, Some(&etag))
+        let etag = write_checkpoint(&store, bucket, context.incarnation_id(), &live, Some(&etag))
             .await
             .expect("live lease");
         assert_eq!(runner.recover_once().await.taken_over, 0, "unexpired lease must not be taken over");
@@ -2162,7 +2350,7 @@ mod tests {
             lease_until: now - Duration::from_secs(1),
         });
         expired.updated_at = now + Duration::from_millis(1);
-        write_checkpoint(&store, bucket, &expired, Some(&etag))
+        write_checkpoint(&store, bucket, context.incarnation_id(), &expired, Some(&etag))
             .await
             .expect("expire lease");
         let stats = runner.recover_once().await;
@@ -2201,7 +2389,7 @@ mod tests {
         crashed.continuation_token = Some("2".to_string());
         crashed.failed = 1;
         crashed.record_failure("local_write", Some("k/00002"), crashed_at);
-        write_checkpoint(&store, bucket, &crashed, None)
+        write_checkpoint(&store, bucket, context.incarnation_id(), &crashed, None)
             .await
             .expect("seed failed page with an expired lease");
 
@@ -2257,7 +2445,9 @@ mod tests {
 
         // Same node name, unexpired lease: only a restart can produce this.
         let own = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-a", now);
-        let etag = write_checkpoint(&store, bucket, &own, None).await.expect("seed");
+        let etag = write_checkpoint(&store, bucket, context.incarnation_id(), &own, None)
+            .await
+            .expect("seed");
         assert_eq!(runner.recover_once().await.taken_over, 1, "own-node running job is reclaimed at once");
         runner.wait_until_idle(bucket).await;
         let cp = read_checkpoint(&store, bucket)
@@ -2275,7 +2465,7 @@ mod tests {
             node: "node-z".to_string(),
             lease_until: now - Duration::from_secs(1),
         });
-        write_checkpoint(&store, bucket, &stale, Some(&stored.etag))
+        write_checkpoint(&store, bucket, context.incarnation_id(), &stale, Some(&stored.etag))
             .await
             .expect("seed stale");
         let stats = runner.recover_once().await;

rustfs/src/on_demand_migration/pull.rs

@@ -243,6 +243,8 @@ impl PullSource for SourceClient {
 #[derive(Clone, Debug)]
 pub struct WriteBackRequest {
     pub bucket: String,
+    /// Identity captured with the source configuration, retained through cleanup.
+    pub bucket_incarnation_id: uuid::Uuid,
     pub key: String,
     /// Source HEAD/GET of the whole object.
     pub head: SourceHead,
@@ -263,6 +265,7 @@ impl WriteBackRequest {
         let config = state.config();
         Self {
             bucket: state.bucket().to_string(),
+            bucket_incarnation_id: state.incarnation_id(),
             key: key.to_string(),
             head,
             source_label: format!("{}:{}", config.source.provider.as_str(), config.source.bucket),
@@ -357,7 +360,7 @@ pub trait OdmWriteBack: Send + Sync {
         parts: Vec<WriteBackPart>,
     ) -> Result<WriteBackOutcome, WriteBackError>;
 
-    async fn abort_multipart_upload(&self, bucket: &str, key: &str, upload_id: &str) -> Result<(), WriteBackError>;
+    async fn abort_multipart_upload(&self, request: &WriteBackRequest, upload_id: &str) -> Result<(), WriteBackError>;
 }
 
 /// Why the pump stopped feeding the write-back before EOF.
@@ -662,9 +665,7 @@ async fn write_multipart(
         Err(err) => Err(err),
     };
     if completed.is_err()
-        && let Err(abort_err) = write_back
+        && let Err(abort_err) = write_back.abort_multipart_upload(request, &upload_id).await
-            .abort_multipart_upload(&request.bucket, &request.key, &upload_id)
-            .await
     {
         debug!(
             event = EVENT_ODM_PULL_FAILED,
@@ -1340,7 +1341,7 @@ mod tests {
             })
         }
 
-        async fn abort_multipart_upload(&self, _bucket: &str, _key: &str, upload_id: &str) -> Result<(), WriteBackError> {
+        async fn abort_multipart_upload(&self, _request: &WriteBackRequest, upload_id: &str) -> Result<(), WriteBackError> {
             self.aborted.lock().push(upload_id.to_string());
             Ok(())
         }

rustfs/src/on_demand_migration/sys.rs

@@ -23,9 +23,8 @@
 //! path (initial load, admin update, peer reload, refresh loop, lazy load).
 //!
 //! Change detection compares the config by value (`PartialEq`) rather than
-//! by `updated_at`: the hook does not carry the timestamp, fetching it would
+//! by `updated_at`. The bucket incarnation is part of this comparison:
-//! re-enter the metadata system from inside its own publish path, and a
+//! recreating a bucket must cancel old work even with identical configuration.
-//! byte-identical config never needs a new client anyway.
 //!
 //! Client construction is async (TLS material may be read from disk), so
 //! the hook does not build inline: `publish` removes state synchronously and
@@ -270,6 +269,7 @@ impl Drop for InflightEntryGuard<'_> {
 /// config change (counters excepted), removed when the config goes away.
 pub struct BucketOdmState {
     bucket: String,
+    incarnation_id: uuid::Uuid,
     config: OnDemandMigrationConfig,
     applied_at: OffsetDateTime,
     endpoint_host: String,
@@ -307,6 +307,7 @@ impl BucketOdmState {
     async fn build(
         bucket: &str,
         config: &OnDemandMigrationConfig,
+        incarnation_id: uuid::Uuid,
         stats: Arc<OdmStats>,
         write_back: Option<Arc<dyn OdmWriteBack>>,
     ) -> Arc<Self> {
@@ -323,6 +324,7 @@ impl BucketOdmState {
         let policy = &config.policy;
         Arc::new(Self {
             bucket: bucket.to_string(),
+            incarnation_id,
             endpoint_host: endpoint_host(&config.source),
             config: config.clone(),
             applied_at: OffsetDateTime::now_utc(),
@@ -340,10 +342,18 @@ impl BucketOdmState {
         })
     }
 
+    pub fn filter_incarnation(self: Arc<Self>, incarnation_id: uuid::Uuid) -> Option<Arc<Self>> {
+        (self.incarnation_id == incarnation_id && !self.is_cancelled()).then_some(self)
+    }
+
     pub fn bucket(&self) -> &str {
         &self.bucket
     }
 
+    pub fn incarnation_id(&self) -> uuid::Uuid {
+        self.incarnation_id
+    }
+
     pub fn config(&self) -> &OnDemandMigrationConfig {
         &self.config
     }
@@ -742,16 +752,17 @@ impl OnDemandMigrationSys {
         BUCKET_CONFIG_PUBLISH_HOOK
             .set(Box::new(move |bucket, config_file, stored| {
                 if config_file == BUCKET_ON_DEMAND_MIGRATION_CONFIG {
-                    self.publish_stored(bucket, stored.map(|(bytes, _)| bytes));
+                    self.publish_stored(bucket, stored.map(|(bytes, _, incarnation)| (bytes, incarnation)));
                 }
             }))
             .is_ok()
     }
 
     /// Corrupt persisted bytes withdraw state synchronously, just like deletion.
-    fn publish_stored(&'static self, bucket: &str, stored: Option<&[u8]>) {
+    fn publish_stored(&'static self, bucket: &str, stored: Option<(&[u8], uuid::Uuid)>) {
-        match stored.map(OnDemandMigrationConfig::from_json).transpose() {
+        let incarnation_id = stored.map(|(_, id)| id).unwrap_or_default();
-            Ok(config) => self.publish(bucket, config.as_ref()),
+        match stored.map(|(bytes, _)| OnDemandMigrationConfig::from_json(bytes)).transpose() {
+            Ok(config) => self.publish_for_incarnation(bucket, incarnation_id, config.as_ref()),
             Err(err) => {
                 warn!(
                     event = EVENT_ODM_BUCKET_STATE_APPLIED,
@@ -762,7 +773,7 @@ impl OnDemandMigrationSys {
                     error = %err,
                     "Failed to parse on-demand migration config"
                 );
-                self.publish(bucket, None);
+                self.publish_for_incarnation(bucket, incarnation_id, None);
             }
         }
     }
@@ -770,14 +781,19 @@ impl OnDemandMigrationSys {
     /// Hook entry point: removals apply immediately, installs are spawned
     /// (client construction is async). Requires a Tokio runtime for the
     /// install path; without one the config is logged and skipped.
-    pub fn publish(&'static self, bucket: &str, config: Option<&OnDemandMigrationConfig>) {
+    pub fn publish_for_incarnation(
-        let config = self.desired(config);
+        &'static self,
+        bucket: &str,
+        incarnation_id: uuid::Uuid,
+        config: Option<&OnDemandMigrationConfig>,
+    ) {
+        let config = self.desired(config).filter(|_| !incarnation_id.is_nil());
         let generation = self.reserve_generation(bucket, config.is_some());
         let Some(config) = config else {
             self.remove_with_generation(bucket, generation);
             return;
         };
-        if self.is_unchanged(bucket, config, generation) {
+        if self.is_unchanged(bucket, incarnation_id, config, generation) {
             return;
         }
         let Ok(handle) = tokio::runtime::Handle::try_current() else {
@@ -796,32 +812,49 @@ impl OnDemandMigrationSys {
         let bucket = bucket.to_string();
         let config = config.clone();
         handle.spawn(async move {
-            self.apply_with_generation(&bucket, Some(&config), generation).await;
+            self.apply_with_generation(&bucket, incarnation_id, Some(&config), generation)
+                .await;
         });
     }
 
     /// Installs, rebuilds, or removes the bucket state for `config`.
     /// Idempotent: the same config on an installed bucket is a no-op.
+    #[cfg(test)]
     pub async fn apply(&self, bucket: &str, config: Option<&OnDemandMigrationConfig>) -> ApplyOutcome {
-        let config = self.desired(config);
+        self.apply_for_incarnation(bucket, uuid::Uuid::from_u128(1), config).await
+    }
+
+    #[cfg(test)]
+    pub fn publish(&'static self, bucket: &str, config: Option<&OnDemandMigrationConfig>) {
+        self.publish_for_incarnation(bucket, uuid::Uuid::from_u128(1), config);
+    }
+
+    pub async fn apply_for_incarnation(
+        &self,
+        bucket: &str,
+        incarnation_id: uuid::Uuid,
+        config: Option<&OnDemandMigrationConfig>,
+    ) -> ApplyOutcome {
+        let config = self.desired(config).filter(|_| !incarnation_id.is_nil());
         let generation = self.reserve_generation(bucket, config.is_some());
-        self.apply_with_generation(bucket, config, generation).await
+        self.apply_with_generation(bucket, incarnation_id, config, generation).await
     }
 
     async fn apply_with_generation(
         &self,
         bucket: &str,
+        incarnation_id: uuid::Uuid,
         config: Option<&OnDemandMigrationConfig>,
         generation: u64,
     ) -> ApplyOutcome {
         let Some(config) = self.desired(config) else {
             return self.remove_with_generation(bucket, generation);
         };
-        if self.is_unchanged(bucket, config, generation) {
+        if self.is_unchanged(bucket, incarnation_id, config, generation) {
             return ApplyOutcome::Unchanged;
         }
         let stats = self.state(bucket).map(|state| Arc::clone(&state.stats)).unwrap_or_default();
-        let state = BucketOdmState::build(bucket, config, stats, self.write_back()).await;
+        let state = BucketOdmState::build(bucket, config, incarnation_id, stats, self.write_back()).await;
 
         let (outcome, previous) = {
             let mut buckets = self.buckets.write();
@@ -871,6 +904,7 @@ impl OnDemandMigrationSys {
 
     /// One-shot lookup: module switch, bucket state, prefix filter,
     /// client availability, negative cache, breaker, in that order.
+    #[cfg(test)]
     pub fn resolve(&self, bucket: &str, key: &str) -> Option<OdmLookup> {
         if !self.is_module_enabled() {
             return None;
@@ -878,6 +912,13 @@ impl OnDemandMigrationSys {
         self.state(bucket)?.resolve_key(key)
     }
 
+    pub fn resolve_for_incarnation(&self, bucket: &str, key: &str, incarnation_id: uuid::Uuid) -> Option<OdmLookup> {
+        if !self.is_module_enabled() {
+            return None;
+        }
+        self.state(bucket)?.filter_incarnation(incarnation_id)?.resolve_key(key)
+    }
+
     pub fn state(&self, bucket: &str) -> Option<Arc<BucketOdmState>> {
         self.buckets.read().get(bucket).and_then(|slot| slot.state.clone())
     }
@@ -925,15 +966,26 @@ impl OnDemandMigrationSys {
 
     /// Claims `generation` for the bucket when the installed state already
     /// matches `config` and has a usable client.
-    fn is_unchanged(&self, bucket: &str, config: &OnDemandMigrationConfig, generation: u64) -> bool {
+    fn is_unchanged(&self, bucket: &str, incarnation_id: uuid::Uuid, config: &OnDemandMigrationConfig, generation: u64) -> bool {
         let mut buckets = self.buckets.write();
         let Some(slot) = buckets.get_mut(bucket) else {
             return false;
         };
+        if slot.generation > generation {
+            return false;
+        }
+        if slot
+            .state
+            .as_ref()
+            .is_some_and(|state| state.incarnation_id != incarnation_id)
+            && let Some(previous) = slot.state.take()
+        {
+            previous.cancel.cancel();
+        }
         let unchanged = slot
             .state
             .as_ref()
-            .is_some_and(|state| state.client.is_ok() && state.config == *config);
+            .is_some_and(|state| state.client.is_ok() && state.incarnation_id == incarnation_id && state.config == *config);
         if unchanged && slot.generation < generation {
             slot.generation = generation;
         }
@@ -1366,13 +1418,88 @@ mod tests {
         assert!(state.is_cancelled());
     }
 
+    #[tokio::test]
+    async fn identical_config_on_recreated_bucket_cancels_old_state() {
+        let sys = enabled_sys();
+        let cfg = config(None);
+        let old_id = uuid::Uuid::new_v4();
+        let new_id = uuid::Uuid::new_v4();
+        sys.apply_for_incarnation("recreated", old_id, Some(&cfg)).await;
+        let old = sys.state("recreated").expect("old state installed");
+        assert!(sys.resolve_for_incarnation("recreated", "key", new_id).is_none());
+        sys.apply_for_incarnation("recreated", new_id, Some(&cfg)).await;
+        let replacement = sys.state("recreated").expect("replacement state installed");
+        assert!(old.is_cancelled());
+        assert!(!Arc::ptr_eq(&old, &replacement));
+        assert_eq!(replacement.incarnation_id(), new_id);
+        assert!(sys.resolve_for_incarnation("recreated", "key", old_id).is_none());
+        assert!(sys.resolve_for_incarnation("recreated", "key", new_id).is_some());
+    }
+
+    #[tokio::test]
+    async fn changed_delete_marker_policy_withdraws_the_captured_lookup() {
+        let sys = enabled_sys();
+        let incarnation = uuid::Uuid::new_v4();
+        let mut cfg = config(None);
+        cfg.policy.respect_local_delete_marker = false;
+        sys.apply_for_incarnation("policy-snapshot", incarnation, Some(&cfg)).await;
+        let captured = sys.state("policy-snapshot").expect("policy A installed");
+        assert!(!captured.config().policy.respect_local_delete_marker);
+
+        cfg.policy.respect_local_delete_marker = true;
+        sys.apply_for_incarnation("policy-snapshot", incarnation, Some(&cfg)).await;
+        let replacement = sys.state("policy-snapshot").expect("policy B installed");
+        assert!(replacement.config().policy.respect_local_delete_marker);
+        assert!(captured.is_cancelled());
+        assert!(
+            captured
+                .filter_incarnation(incarnation)
+                .and_then(|state| state.resolve_key("key"))
+                .is_none(),
+            "a request that evaluated policy A cannot continue through policy B"
+        );
+        assert!(
+            replacement
+                .clone()
+                .filter_incarnation(incarnation)
+                .and_then(|state| state.resolve_key("key"))
+                .is_some()
+        );
+        assert_eq!(
+            replacement
+                .stats()
+                .snapshot(replacement.breaker().state())
+                .source_latency
+                .count,
+            0
+        );
+    }
+
+    #[tokio::test]
+    async fn missing_incarnation_cannot_install_or_retain_source_state() {
+        let sys: &'static OnDemandMigrationSys = Box::leak(Box::new(enabled_sys()));
+        let cfg = config(None);
+        assert_eq!(
+            sys.apply_for_incarnation("missing", uuid::Uuid::nil(), Some(&cfg)).await,
+            ApplyOutcome::NotDesired
+        );
+        sys.publish_for_incarnation("missing", uuid::Uuid::nil(), Some(&cfg));
+        assert!(sys.state("missing").is_none());
+
+        sys.apply_for_incarnation("missing", uuid::Uuid::new_v4(), Some(&cfg)).await;
+        let state = sys.state("missing").expect("valid identity installed");
+        sys.publish_for_incarnation("missing", uuid::Uuid::nil(), Some(&cfg));
+        assert!(sys.state("missing").is_none());
+        assert!(state.is_cancelled());
+    }
+
     #[tokio::test]
     async fn corrupt_stored_config_withdraws_runtime_state() {
         let sys: &'static OnDemandMigrationSys = Box::leak(Box::new(enabled_sys()));
         let cfg = config(None);
         assert_eq!(sys.apply("corrupt", Some(&cfg)).await, ApplyOutcome::Installed);
         let state = sys.state("corrupt").expect("state installed");
-        sys.publish_stored("corrupt", Some(b"not-json"));
+        sys.publish_stored("corrupt", Some((b"not-json", uuid::Uuid::from_u128(1))));
         assert!(sys.state("corrupt").is_none(), "corruption cannot keep an older source active");
         assert!(state.is_cancelled(), "corruption cancels in-flight work");
     }
@@ -1395,7 +1522,11 @@ mod tests {
         let older = sys.reserve_generation("b", true);
         let newer = sys.reserve_generation("b", false);
         assert_eq!(sys.remove_with_generation("b", newer), ApplyOutcome::NotDesired);
-        assert_eq!(sys.apply_with_generation("b", Some(&cfg), older).await, ApplyOutcome::Superseded);
+        assert_eq!(
+            sys.apply_with_generation("b", uuid::Uuid::from_u128(1), Some(&cfg), older)
+                .await,
+            ApplyOutcome::Superseded
+        );
         assert!(sys.state("b").is_none(), "removal must supersede an in-flight first install");
 
         assert_eq!(sys.apply("b", Some(&cfg)).await, ApplyOutcome::Installed);
@@ -1404,7 +1535,11 @@ mod tests {
         let newer = sys.reserve_generation("b", false);
         assert_eq!(sys.remove_with_generation("b", newer), ApplyOutcome::Removed);
         assert!(installed.is_cancelled());
-        assert_eq!(sys.apply_with_generation("b", Some(&cfg), older).await, ApplyOutcome::Superseded);
+        assert_eq!(
+            sys.apply_with_generation("b", uuid::Uuid::from_u128(1), Some(&cfg), older)
+                .await,
+            ApplyOutcome::Superseded
+        );
         assert!(sys.state("b").is_none(), "the stale install is discarded");
     }
 

rustfs/src/storage/access.rs

@@ -288,6 +288,68 @@ async fn load_bucket_generation<T>(fs: &FS, req: &S3Request<T>, bucket: &str) ->
     load_bucket_generation_from_store(store.as_ref(), req, bucket).await
 }
 
+/// A read may consult only the source admitted before its authorization.
+/// Capture failures are deferred until a local miss actually needs that source.
+#[derive(Clone, Debug)]
+enum OdmReadGenerationGuard {
+    Unavailable,
+    Ready(BucketGenerationGuard),
+    Failed { code: S3ErrorCode, message: String },
+}
+
+impl OdmReadGenerationGuard {
+    fn from_result(result: S3Result<BucketGenerationGuard>) -> Self {
+        match result {
+            Ok(guard) => Self::Ready(guard),
+            Err(err) => Self::Failed {
+                code: err.code().clone(),
+                message: err.message().unwrap_or_else(|| err.code().as_str()).to_string(),
+            },
+        }
+    }
+}
+
+fn odm_read_source_configured(bucket: &str) -> bool {
+    let sys = crate::on_demand_migration::OnDemandMigrationSys::get();
+    sys.is_module_enabled() && sys.state(bucket).is_some()
+}
+
+async fn capture_odm_read_generation<T>(fs: &FS, req: &S3Request<T>, bucket: &str) -> OdmReadGenerationGuard {
+    if !odm_read_source_configured(bucket) {
+        return OdmReadGenerationGuard::Unavailable;
+    }
+    OdmReadGenerationGuard::from_result(load_bucket_generation(fs, req, bucket).await)
+}
+
+/// Direct usecase callers have no access middleware; capture at their entry.
+/// A server request without the access marker must never bind to a later source.
+pub(crate) async fn prepare_odm_read_generation<T>(
+    store: &crate::storage::storage_api::ECStore,
+    req: &mut S3Request<T>,
+    bucket: &str,
+) {
+    if req.extensions.get::<OdmReadGenerationGuard>().is_some() {
+        return;
+    }
+    let guard = if req.extensions.get::<std::sync::Arc<ServerContextSlot>>().is_some() || !odm_read_source_configured(bucket) {
+        OdmReadGenerationGuard::Unavailable
+    } else {
+        OdmReadGenerationGuard::from_result(load_bucket_generation_from_store(store, req, bucket).await)
+    };
+    req.extensions.insert(guard);
+}
+
+pub(crate) fn odm_read_generation<T>(req: &S3Request<T>, bucket: &str) -> S3Result<Option<uuid::Uuid>> {
+    match req.extensions.get::<OdmReadGenerationGuard>() {
+        Some(OdmReadGenerationGuard::Ready(guard)) if guard.bucket == bucket => Ok(Some(guard.incarnation_id)),
+        Some(OdmReadGenerationGuard::Ready(_)) => {
+            Err(s3_error!(InternalError, "source generation guard does not match request bucket"))
+        }
+        Some(OdmReadGenerationGuard::Failed { code, message }) => Err(S3Error::with_message(code.clone(), message.clone())),
+        Some(OdmReadGenerationGuard::Unavailable) | None => Ok(None),
+    }
+}
+
 async fn load_copy_source_bucket_generation(fs: &FS, bucket: &str) -> S3Result<CopySourceBucketGenerationGuard> {
     let store = fs
         .server_ctx()
@@ -2365,6 +2427,8 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn get_object(&self, req: &mut S3Request<GetObjectInput>) -> S3Result<()> {
+        let bucket = req.input.bucket.clone();
+        let source_generation = capture_odm_read_generation(self, req, &bucket).await;
         let req_info = ext_req_info_mut(&mut req.extensions)?;
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = Some(req.input.key.clone());
@@ -2372,7 +2436,9 @@ impl S3Access for FS {
 
         // GHSA-3ppv: a versioned read (?versionId=...) must authorize against
         // s3:GetObjectVersion, not s3:GetObject.
-        authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await
+        authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await?;
+        req.extensions.insert(source_generation);
+        Ok(())
     }
 
     /// Checks whether the GetObjectAcl request has accesses to the resources.
@@ -2484,6 +2550,8 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn head_object(&self, req: &mut S3Request<HeadObjectInput>) -> S3Result<()> {
+        let bucket = req.input.bucket.clone();
+        let source_generation = capture_odm_read_generation(self, req, &bucket).await;
         let req_info = ext_req_info_mut(&mut req.extensions)?;
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = Some(req.input.key.clone());
@@ -2496,10 +2564,13 @@ impl S3Access for FS {
         if get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_CHECK).as_deref() == Some("true") {
             authorize_request(req, Action::S3Action(S3Action::ReplicateObjectAction)).await?;
             req_info_mut(req)?.replication_request_authorized = true;
+            req.extensions.insert(source_generation);
             return Ok(());
         }
 
-        authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await
+        authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await?;
+        req.extensions.insert(source_generation);
+        Ok(())
     }
 
     /// Checks whether the ListBucketAnalyticsConfigurations request has accesses to the resources.
@@ -2588,10 +2659,14 @@ impl S3Access for FS {
     ///
     /// This method returns `Ok(())` by default.
     async fn list_objects_v2(&self, req: &mut S3Request<ListObjectsV2Input>) -> S3Result<()> {
+        let bucket = req.input.bucket.clone();
+        let source_generation = capture_odm_read_generation(self, req, &bucket).await;
         let req_info = ext_req_info_mut(&mut req.extensions)?;
         req_info.bucket = Some(req.input.bucket.clone());
 
-        authorize_request(req, Action::S3Action(S3Action::ListBucketAction)).await
+        authorize_request(req, Action::S3Action(S3Action::ListBucketAction)).await?;
+        req.extensions.insert(source_generation);
+        Ok(())
     }
 
     /// Checks whether the ListParts request has accesses to the resources.
@@ -3858,6 +3933,285 @@ mod tests {
         assert_eq!(req_info.object.as_deref(), Some("test-key"));
     }
 
+    #[test]
+    #[serial]
+    fn odm_read_capture_preserves_access_and_parameter_error_order() {
+        crate::app::gating_test_env::run_large_stack_test("odm-access-order", || async {
+            let context = crate::app::gating_test_env::shared_gating_ambient().await;
+            let server_ctx = ServerContextSlot::new();
+            assert!(server_ctx.install(Arc::clone(&context)));
+            let fs = FS::with_server_ctx(server_ctx);
+            let sys = crate::on_demand_migration::OnDemandMigrationSys::get();
+            let enabled_before = sys.is_module_enabled();
+            let bucket = format!("odm-access-missing-{}", uuid::Uuid::new_v4());
+            for enabled in [false, true] {
+                sys.set_module_enabled(enabled);
+                let mut get = build_request(
+                    GetObjectInput {
+                        bucket: bucket.clone(),
+                        key: "key".into(),
+                        part_number: Some(0),
+                        ..Default::default()
+                    },
+                    Method::GET,
+                );
+                get.extensions.insert(ReqInfo {
+                    cred: Some(rustfs_credentials::Credentials::default()),
+                    is_owner: true,
+                    ..Default::default()
+                });
+                get.extensions.insert(fs.server_ctx().clone());
+                let mut head = get.clone().map_input(|_| HeadObjectInput {
+                    bucket: bucket.clone(),
+                    key: "key".into(),
+                    part_number: Some(1),
+                    range: Some(s3s::dto::Range::Int { first: 0, last: Some(1) }),
+                    ..Default::default()
+                });
+                head.method = Method::HEAD;
+                let mut list = get.clone().map_input(|_| ListObjectsV2Input {
+                    bucket: bucket.clone(),
+                    max_keys: Some(-1),
+                    ..Default::default()
+                });
+                fs.get_object(&mut get)
+                    .await
+                    .expect("ordinary GET access must not require bucket identity");
+                fs.head_object(&mut head)
+                    .await
+                    .expect("ordinary HEAD access must not require bucket identity");
+                fs.list_objects_v2(&mut list)
+                    .await
+                    .expect("ordinary LIST access must not require bucket identity");
+                assert!(matches!(
+                    get.extensions.get::<super::OdmReadGenerationGuard>(),
+                    Some(super::OdmReadGenerationGuard::Unavailable)
+                ));
+                assert!(matches!(
+                    head.extensions.get::<super::OdmReadGenerationGuard>(),
+                    Some(super::OdmReadGenerationGuard::Unavailable)
+                ));
+                assert!(matches!(
+                    list.extensions.get::<super::OdmReadGenerationGuard>(),
+                    Some(super::OdmReadGenerationGuard::Unavailable)
+                ));
+                let usecase = crate::app::object_usecase::DefaultObjectUsecase::with_context(Some(Arc::clone(&context)));
+                assert_eq!(
+                    usecase.execute_get_object(get).await.expect_err("bad GET part number").code(),
+                    &S3ErrorCode::InvalidArgument
+                );
+                assert_eq!(
+                    usecase
+                        .execute_head_object(head)
+                        .await
+                        .expect_err("range and part number conflict")
+                        .code(),
+                    &S3ErrorCode::InvalidArgument
+                );
+                assert_eq!(
+                    crate::app::bucket_usecase::DefaultBucketUsecase::with_context(Some(Arc::clone(&context)))
+                        .execute_list_objects_v2(list)
+                        .await
+                        .expect_err("negative max keys")
+                        .code(),
+                    &S3ErrorCode::InvalidArgument
+                );
+            }
+            sys.set_module_enabled(enabled_before);
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn odm_read_capture_failure_is_deferred_until_source_miss() {
+        crate::app::gating_test_env::run_large_stack_test("odm-access-failure", || async {
+            let context = crate::app::gating_test_env::shared_gating_ambient().await;
+            let store = context.object_store();
+            let server_ctx = ServerContextSlot::new();
+            assert!(server_ctx.install(Arc::clone(&context)));
+            let fs = FS::with_server_ctx(server_ctx);
+            let sys = crate::on_demand_migration::OnDemandMigrationSys::get();
+            let enabled_before = sys.is_module_enabled();
+            sys.set_module_enabled(true);
+            let bucket = format!("odm-capture-failure-{}", uuid::Uuid::new_v4());
+            let mut config: crate::on_demand_migration::OnDemandMigrationConfig = serde_json::from_str(r#"{"source":{"provider":"minio","endpoint":"https://source.example.com","bucket":"source","credentials":{"access_key":"test","secret_key":"test"}}}"#).expect("source config");
+            config.policy.list_through = true;
+            sys.apply_for_incarnation(&bucket, uuid::Uuid::new_v4(), Some(&config)).await;
+            let mut get = build_request(
+                GetObjectInput {
+                    bucket: bucket.clone(),
+                    key: "local".into(),
+                    ..Default::default()
+                },
+                Method::GET,
+            );
+            get.extensions.insert(ReqInfo {
+                cred: Some(rustfs_credentials::Credentials::default()),
+                is_owner: true,
+                ..Default::default()
+            });
+            get.extensions.insert(fs.server_ctx().clone());
+            let mut head = get.clone().map_input(|_| HeadObjectInput {
+                bucket: bucket.clone(),
+                key: "local".into(),
+                ..Default::default()
+            });
+            head.method = Method::HEAD;
+            let mut list = get.clone().map_input(|_| ListObjectsV2Input {
+                bucket: bucket.clone(),
+                ..Default::default()
+            });
+            fs.list_objects_v2(&mut list)
+                .await
+                .expect("capture failure must not preempt LIST authorization");
+            fs.get_object(&mut get)
+                .await
+                .expect("capture failure must not preempt GET authorization");
+            fs.head_object(&mut head)
+                .await
+                .expect("capture failure must not preempt HEAD authorization");
+            assert!(matches!(
+                get.extensions.get::<super::OdmReadGenerationGuard>(),
+                Some(super::OdmReadGenerationGuard::Failed { .. })
+            ));
+            assert!(matches!(
+                head.extensions.get::<super::OdmReadGenerationGuard>(),
+                Some(super::OdmReadGenerationGuard::Failed { .. })
+            ));
+            store
+                .make_bucket(
+                    &bucket,
+                    &MakeBucketOptions {
+                        versioning_enabled: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("create bucket after capture");
+            store
+                .put_object(
+                    &bucket,
+                    "local",
+                    &mut crate::storage::PutObjReader::from_vec(b"local".to_vec()),
+                    &crate::storage::ObjectOptions::default(),
+                )
+                .await
+                .expect("create local hit");
+            // Publishing a usable source later must not repair a failed capture.
+            let incarnation = store.bucket_incarnation_id(&bucket).await.expect("created identity");
+            sys.apply_for_incarnation(&bucket, incarnation, Some(&config)).await;
+            let usecase = crate::app::object_usecase::DefaultObjectUsecase::with_context(Some(Arc::clone(&context)));
+            usecase
+                .execute_get_object(get.clone())
+                .await
+                .expect("a local GET hit ignores source capture failure");
+            usecase
+                .execute_head_object(head.clone())
+                .await
+                .expect("a local HEAD hit ignores source capture failure");
+            get.input.key = "missing".into();
+            head.input.key = "missing".into();
+            assert_eq!(
+                usecase
+                    .execute_get_object(get.clone())
+                    .await
+                    .expect_err("failed capture cannot rebind on GET miss")
+                    .code(),
+                &S3ErrorCode::NoSuchBucket
+            );
+            assert_eq!(
+                usecase
+                    .execute_head_object(head.clone())
+                    .await
+                    .expect_err("failed capture cannot rebind on HEAD miss")
+                    .code(),
+                &S3ErrorCode::NoSuchBucket
+            );
+            assert_eq!(
+                crate::app::bucket_usecase::DefaultBucketUsecase::with_context(Some(Arc::clone(&context)))
+                    .execute_list_objects_v2(list.clone())
+                    .await
+                    .expect_err("failed capture cannot rebind on LIST")
+                    .code(),
+                &S3ErrorCode::NoSuchBucket
+            );
+            config.filter.prefix = Some("remote/".into());
+            sys.apply_for_incarnation(&bucket, incarnation, Some(&config)).await;
+            assert_eq!(
+                usecase
+                    .execute_get_object(get.clone())
+                    .await
+                    .expect_err("filtered GET remains local")
+                    .code(),
+                &S3ErrorCode::NoSuchKey
+            );
+            assert_eq!(
+                usecase
+                    .execute_head_object(head.clone())
+                    .await
+                    .expect_err("filtered HEAD remains local")
+                    .code(),
+                &S3ErrorCode::NoSuchKey
+            );
+            list.input.prefix = Some("local/".into());
+            let listing = crate::app::bucket_usecase::DefaultBucketUsecase::with_context(Some(Arc::clone(&context)))
+                .execute_list_objects_v2(list.clone())
+                .await
+                .expect("disjoint prefix needs no source identity");
+            assert_eq!(listing.output.key_count, Some(0));
+            list.input.prefix = Some("remote/".into());
+            list.input.max_keys = Some(0);
+            let listing = crate::app::bucket_usecase::DefaultBucketUsecase::with_context(Some(Arc::clone(&context)))
+                .execute_list_objects_v2(list)
+                .await
+                .expect("an empty page needs no source identity");
+            assert_eq!(listing.output.key_count, Some(0));
+
+            config.filter.prefix = None;
+            config.policy.head = crate::on_demand_migration::HeadPolicy::LocalOnly;
+            sys.apply_for_incarnation(&bucket, incarnation, Some(&config)).await;
+            assert_eq!(
+                usecase
+                    .execute_head_object(head.clone())
+                    .await
+                    .expect_err("local-only HEAD needs no source identity")
+                    .code(),
+                &S3ErrorCode::NoSuchKey
+            );
+            config.policy.head = crate::on_demand_migration::HeadPolicy::Proxy;
+            sys.apply_for_incarnation(&bucket, incarnation, Some(&config)).await;
+            store
+                .delete_object(
+                    &bucket,
+                    "missing",
+                    crate::storage::ObjectOptions {
+                        versioned: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("create local delete marker");
+            assert_eq!(
+                usecase
+                    .execute_get_object(get)
+                    .await
+                    .expect_err("GET respects the delete marker before capture failure")
+                    .code(),
+                &S3ErrorCode::NoSuchKey
+            );
+            assert_eq!(
+                usecase
+                    .execute_head_object(head)
+                    .await
+                    .expect_err("HEAD respects the delete marker before capture failure")
+                    .code(),
+                &S3ErrorCode::NoSuchKey
+            );
+            sys.remove(&bucket);
+            sys.set_module_enabled(enabled_before);
+        });
+    }
+
     #[tokio::test]
     #[serial]
     async fn put_object_access_captures_authorized_bucket_incarnation() {

rustfs/src/storage/storage_api.rs

@@ -119,8 +119,9 @@ pub(crate) mod access_consumer {
     pub(crate) use super::super::access::{
         PostObjectRequestMarker, ReqInfo, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard,
         authorize_internal_object_request, authorize_request, bucket_config_mutation_incarnation, has_bypass_governance_header,
-        load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, prepare_list_buckets_iam_authorization,
+        load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation,
-        recursive_force_delete_is_authorized, replication_request_authorized, req_info_mut, req_info_ref,
+        prepare_list_buckets_iam_authorization, prepare_odm_read_generation, recursive_force_delete_is_authorized,
+        replication_request_authorized, req_info_mut, req_info_ref,
     };
 }
 

rustfs/src/storage_api.rs

@@ -430,10 +430,12 @@ pub(crate) mod on_demand_migration {
 
     #[cfg(test)]
     pub(crate) mod test_support {
+        pub(crate) use crate::storage::storage_api::contract::bucket::{BUCKET_LIFECYCLE_LOCK_OBJECT, BucketOperations};
         pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata::{
             BUCKET_ON_DEMAND_MIGRATION_CONFIG, BucketMetadata,
         };
-        pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::BucketMetadataSys;
         pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::test_support::isolated_store_over_temp_disks;
+        pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::{BucketMetadataSys, init_bucket_metadata_sys};
+        pub(crate) use crate::storage::storage_api::ecstore_set_disk::{PutObjectCommitBarrier, PutObjectCommitPause};
     }
 }