fix: address rc.1 release blockers (#5648)

* fix: address rc.1 release blockers

* fix: route release guards through architecture boundaries

* fix: close remaining rc.1 regression gaps

* refactor: group multipart listing options

* fix: resolve rc.1 CI regressions

* fix(ecstore): keep bucket-config writes off the caller's stack

A bucket-config write nests incarnation resolution (which can drive legacy
migration and a peer fan-out), a full metadata load, and `save` — itself an
object PUT that pulls in the whole erasure write path. Every request that
mutates bucket config is already several futures deep, so inlining all of
that into one state machine overflows the 2MiB worker stack in debug builds.

Two CI lanes aborted with SIGABRT on this:

  ILM Integration (serial)
    rustfs app::lifecycle_transition_api_test::
      compensation_driven_complete_multipart_upload_still_transitions
  Test and Lint (swift)
    rustfs-protocols::swift_metadata_persistence::
      swift_metadata_writes_are_durable

Neither test file is touched by this branch and both lanes are green on
main. Stack-pointer probing showed ~780KiB consumed between
`metadata_sys::update` and the config read alone, with single hops of
363KiB (`update` -> `acquire_config_write_guard_for_incarnation`), 125KiB
and 105KiB.

Box the deep sub-futures on both read-modify-write paths (`update` /
`update_checked` and `update_config_with` / `update_config_with_checked`)
so each guard's own state machine stays small. Behaviour is unchanged;
`update` -> guard drops to 253KiB and both tests pass on the default stack.

* fix(lifecycle): unbreak restore under the bucket generation fence

The ILM lane aborted on a stack overflow before reaching these, so they
were never reported; with that fixed, four restore tests fail. All four
are green on main and none of their test files are touched by this branch.

1. RestoreObject and ListMultipartUploads hard-required
   `opts.expected_bucket_incarnation_id`, but `apply_bucket_generation_guard`
   deliberately leaves it unset when no guard extension is present — only the
   S3 access layer installs one. Every direct caller therefore got
   `InternalError: ... bucket generation guard is missing`. Resolve the
   current generation instead, the way the copy path already does. The fence
   is unaffected: RestoreObject still re-reads the incarnation from disk and
   compares before admitting the restore, and the multipart listing is
   filtered by the value it resolves.

2. `restore_expiry_snapshot_matches` (new on this branch) rejected every
   restored-copy expiry whose `restore_expires` had not already elapsed.
   Whether the restored copy is due to expire is the ILM evaluator's
   decision, made when it emitted DeleteRestoredAction; re-deriving it in
   the set layer only adds a way for a legitimate action to be rejected.
   The stale-event risk it appears to guard is already covered by the
   surrounding snapshot match — a re-restore rewrites `restore_expires`,
   so a replayed event fails the equality check. Drop the clause; the
   fifteen identity clauses are unchanged.

Fixed:
  rustfs app::lifecycle_transition_api_test::
    restore_object_usecase_accepts_exactly_one_of_two_concurrent_restores
    restore_object_usecase_completes_suspended_null_version_in_place
    restore_object_usecase_reports_ongoing_conflict
  rustfs-scanner::lifecycle_integration_test serial_tests::
    test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore

Verification: the CI ILM lane filter now runs 53/53 green locally.

* chore: address review follow-ups on this branch

Four items from the adversarial review that were still open.

- Restore the assertion `test_bucket_replication_replayed_delete_marker_
  preserves_source_mtime_without_source_restart` is named for. The branch
  had replaced the backlog#867 mtime check with `assert_replication_
  converged`, which any successful replication satisfies, and deleted the
  two helpers it needed — so the regression the test exists to catch would
  now pass. This matters here specifically because the branch changes the
  flag feeding `replication_delete_remove_options` and routes replay
  through a new file and ordering.

- Drop `read_config_no_lock_preserve_empty`: zero production callers (the
  one real consumer calls the `_with_metadata` variant directly). Its test
  stanza now exercises that variant, so the coverage moves to live code
  rather than being deleted.

- Revert the `bytesize` bump. It is a no-op: `Cargo.lock` already pinned
  2.7.0 before this branch and is untouched, so the caret range already
  resolved there. Nothing in the diff uses the crate.

- Split the AGENTS.md "Adversarial Validation" policy change out of this
  branch. The edit is defensible on its own, but it relaxes the review gate
  that this branch has to pass, so it should land as its own PR reviewed on
  its own merits rather than bundled with the change that benefits from it.
  The reverted hunks are unchanged and ready to re-apply.

Not changed, deliberately: the missing-sidecar path still fails closed.
`missing_bucket_incarnation_sidecar_for_new_metadata_fails_closed` pins
that on purpose, and serving a non-authoritative Object Lock state would
be the wrong trade. The residual concern stands and is recorded in review
— a crash between the two writes in `persist_new_and_set` leaves the
bucket unloadable until DeleteBucket+CreateBucket, and the repair branches
in `migrate_legacy_metadata` and `make_bucket` are unreachable dead code
for that case. Resolving it needs the read path and the (transaction-lock
holding) repair path to be separated, which is more than a follow-up edit.

* test(ci): serialize the new bucket-incarnation tests

The five tests this branch adds around the incarnation / lifecycle fence
drive `init_bucket_metadata_sys` and `bucket_metadata_sys_of` — process-global
OnceLock state that `serial_test`'s `#[serial]` cannot protect across
nextest's process boundary — and they delete+recreate buckets, the shape that
raced into InsufficientWriteQuorum in backlog#937.

Add them to the `ecstore-serial-flaky` group in both the default and ci
profiles (nextest evaluates a named profile's own overrides list, so the
ci mirror is required). Preventive serialization only, no retries.

Not a full fix for the review comment: `bucket_delete_waits_for_config_
mutation_fence` still proves liveness with a fixed 200ms sleep plus
`assert!(!delete.is_finished())`. Turning that into readiness polling needs
a production-side signal to wait on — asserting "still blocked" is inherently
a negative. Serializing the group removes the parallel-load pressure that
makes the window fragile; the sleep itself is left for a follow-up.

* test(ecstore): pin that a drained bucket is actually deletable

`DeleteBucket`'s emptiness check is `has_xlmeta_files`, a raw scan of the
bucket directory on local disks — not an S3-level listing. So "the client
drained the bucket" and "the bucket is deletable" are two different
contracts, and only the first one was covered.

That gap is what the `S3 Implemented Tests` lane is failing on: 219 cases,
all `BucketNotEmpty` on `nuke_prefixed_buckets`, with every test body
passing. The first one is `test_versioning_obj_suspend_versions`, reported
by pytest as PASSED followed by ERROR at teardown.

Add the missing assertion for the unversioned path: PUT, client DELETE,
then assert no `xl.meta` survives and `DeleteBucket` succeeds. It passes —
which is itself a result: the plain delete path leaves no residue, so the
s3-tests failure is not there.

The versioning-suspended path is the remaining suspect (the client DELETE
leaves a null delete marker, and draining means purging it by
`versionId=null`). It is not covered here: `BucketVersioningSys` resolves
through the ambient `get_bucket_metadata_sys()` OnceLock, which this unit
env cannot set, so the bucket never actually reports as suspended. That
repro belongs at the e2e layer where a real server owns the versioning
state.

* fix(ecstore): let an explicit null-version delete purge its delete marker

Root cause of the `S3 Implemented Tests` lane: 219 cases, all
`BucketNotEmpty` on `nuke_prefixed_buckets`, every test body passing.

On a versioning-suspended bucket a client DELETE leaves a null delete
marker — correct S3 semantics, and an `xl.meta` on disk. Draining the
bucket therefore means purging that marker as `?versionId=null`, which is
what `nuke_bucket` does before `DeleteBucket`. That purge was rejected:

    explicit null-version purge of the null delete marker must succeed,
    got [Some(MethodNotAllowed)]

so the marker survived, and `DeleteBucket`'s emptiness check — a raw
`has_xlmeta_files` scan of the bucket directory, not an S3 listing — kept
reporting the bucket as non-empty.

The two sides of the version comparison in the batch delete loop are in
different namespaces. `goi.version_id` is the client-facing identity, where
`from_file_info` synthesizes `Some(Uuid::nil())` for a null version on a
versioned *or versioning-suspended* bucket. `version_id` is the storage
identity, where `delete_file_info_version_id` maps an explicit
`?versionId=null` to `None`. Comparing them raw makes the purge look like a
version mismatch, so `explicit_delete_marker` is false and the
`MethodNotAllowed` from the lookup is recorded as a delete failure.

This only became reachable on this branch: previously `check_opts` did not
carry `dobj.version_id`, so `set_disk_delete_creates_delete_marker` was
true, `object_lock_check_required` was false, and the lookup that produces
`MethodNotAllowed` never ran. Adding the version id to `check_opts` lit up
a comparison that was already wrong.

Normalize both sides through `delete_file_info_version_id`.

The regression test injects a real Suspended bucket-config snapshot — the
delete path reads versioned/suspended from that snapshot, not from `opts`,
so without it `from_file_info` never synthesizes the null version id and
the branch is not reached. Mutation-checked: restoring the raw comparison
fails the test with the exact `MethodNotAllowed` above.

* fix(app): drop the now-needless struct update

Reverting `crates/replication` to main removed the extra `MrfReplicateEntry`
fields, so this literal specifies every field again and `..Default::default()`
trips `clippy::needless_update` under `-D warnings`.

Caught by CI, not locally: I had run `cargo check --workspace --all-targets`,
which does not see clippy-only lints. Ran `cargo clippy --workspace
--all-targets -- -D warnings` here — clean.

* test(e2e): assert the fresh-volume classification

four_node_empty_legacy_volumes_start_as_fresh only started the cluster and
listed buckets — no assertion, so any classification path that still permits
startup left it green without proving the pre-created empty `.minio.sys`
directories were treated as fresh volumes.

Pin what that classification actually leaves behind: no buckets adopted into
the namespace, `.rustfs.sys/format.json` written on every drive, and the empty
legacy directory left untouched rather than migrated into.

* fix(bucket): apply the requested Object Lock to existing buckets

Site replication replays make-with-versioning against the destination,
carrying the source's `lockEnabled`. When the destination bucket already
exists it takes `force_create`, and the whole option-application block was
gated on `confirmed_missing` — so the call returned success while the replica
stayed unlocked. Replicated versions could then be deleted without the
retention the source enforces.

Object Lock enable is one-way, so applying it to an existing bucket is safe:
move it out of the creation-only gate, keeping `created` and versioning-only
options creation-scoped as before.

An existing authoritative bucket takes the `cache_bucket_metadata_in` branch,
which only caches, so the enable would have been dropped on restart. Persist
instead when the enable actually changed something.

Mutation-checked: restoring the creation-only gate fails the new
`force_create_enables_object_lock_on_an_existing_bucket` with "Object Lock
must be enabled on the existing bucket".

cargo nextest run -p rustfs-ecstore --lib: 3633 passed.

* fix(ecstore): box the generation-checked config mutation paths too

The earlier stack fix boxed `update` and `delete`, but an authorized
bucket-config mutation carrying an incarnation takes `update_if_incarnation`
/ `delete_if_incarnation` instead — which were still inlining the whole
resolve/load/save chain into an already-deep request future. Same overflow,
sibling path.

* fix(restore): keep the nil-version normalization the strip removed

Reverting the replication subsystem to main took `set_disk/replication.rs`
with it, but one line in that file was this branch's own fix rather than
replication work:

    -  self.version_id.filter(|v| !v.is_nil()) == fi.version_id.filter(|v| !v.is_nil())
    +  self.version_id == fi.version_id

For a versioning-suspended object the expected version is `Some(Uuid::nil())`
while the read-back `FileInfo` carries `None`, so the raw compare reports
every suspended restore as "restored object changed before restore metadata
finalization" and the copy-back never commits. Same nil-vs-None mismatch as
the null delete-marker purge fixed earlier on this branch.

Caught by `Test and Lint (rio-v2)`, not by my local runs: the test lives in
`transition_commit_failure_tests`, gated behind `feature = "test-util"`, so
the 3633-test suite I had been running never included it. Re-ran with
`--features rio-v2,test-util`: 3722 passed.

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -3791,6 +3791,29 @@ impl SetDisks {
         Ok(m)
     }
 
+    fn reduce_delete_prefix_results(results: Vec<disk::error::Result<()>>, write_quorum: usize) -> disk::error::Result<()> {
+        let has_existing_volume = results
+            .iter()
+            .any(|result| matches!(result, Ok(()) | Err(DiskError::FileNotFound)));
+        let volume_not_found_count = results
+            .iter()
+            .filter(|result| matches!(result, Err(DiskError::VolumeNotFound)))
+            .count();
+        let errs = results
+            .into_iter()
+            .map(|result| result.err().filter(|err| !DiskError::is_err_object_not_found(err)))
+            .collect::<Vec<_>>();
+
+        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+            return Err(err);
+        }
+        if !has_existing_volume && volume_not_found_count >= write_quorum {
+            return Err(DiskError::VolumeNotFound);
+        }
+
+        Ok(())
+    }
+
     pub(in crate::set_disk) async fn delete_prefix(&self, bucket: &str, prefix: &str) -> disk::error::Result<()> {
         let disks = self.get_disks_internal().await;
         let write_quorum = disks.len() / 2 + 1;
@@ -3813,18 +3836,12 @@ impl SetDisks {
                     )
                     .await
                 } else {
-                    Ok(())
+                    Err(DiskError::DiskNotFound)
                 }
             });
         }
 
-        let errs = join_all(futures).await.into_iter().map(|v| v.err()).collect::<Vec<_>>();
-
-        if let Some(err) = reduce_write_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, write_quorum) {
-            return Err(err);
-        }
-
-        Ok(())
+        Self::reduce_delete_prefix_results(join_all(futures).await, write_quorum)
     }
 
     /// Scan a single disk's copy of `prefix` and decide whether it is an orphan
@@ -5744,16 +5761,147 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn delete_prefix_removes_present_disks_and_ignores_missing_disk_slots() {
+    async fn delete_prefix_succeeds_when_present_disks_reach_quorum() {
         let bucket = "delete-prefix-bucket";
-        let (_dir, disk) = read_multiple_test_disk(bucket, &[("prefix/object.txt", b"payload".as_slice())]).await;
-        let set = io_primitives_test_set(vec![Some(disk.clone()), None], 1).await;
+        let (_dir1, disk1) = read_multiple_test_disk(bucket, &[("prefix/object.txt", b"one".as_slice())]).await;
+        let (_dir2, disk2) = read_multiple_test_disk(bucket, &[("prefix/object.txt", b"two".as_slice())]).await;
+        let (_dir3, disk3) = read_multiple_test_disk(bucket, &[("prefix/object.txt", b"three".as_slice())]).await;
+        let set = io_primitives_test_set(vec![Some(disk1.clone()), Some(disk2.clone()), Some(disk3.clone()), None], 2).await;
 
         set.delete_prefix(bucket, "prefix")
             .await
-            .expect("missing disk slots should not block prefix deletion");
+            .expect("three successful disks should meet a four-disk write quorum");
 
-        assert!(matches!(disk.read_all(bucket, "prefix/object.txt").await, Err(DiskError::FileNotFound)));
+        for disk in [disk1, disk2, disk3] {
+            assert!(matches!(disk.read_all(bucket, "prefix/object.txt").await, Err(DiskError::FileNotFound)));
+        }
+    }
+
+    #[tokio::test]
+    async fn delete_prefix_counts_confirmed_absence_toward_quorum() {
+        let bucket = "delete-prefix-confirmed-absence";
+        let (_dir1, disk1) = read_multiple_test_disk(bucket, &[("prefix/object.txt", b"one".as_slice())]).await;
+        let (_dir2, disk2) = read_multiple_test_disk(bucket, &[("prefix/object.txt", b"two".as_slice())]).await;
+        let (_dir3, disk3) = read_multiple_test_disk(bucket, &[]).await;
+        let (_dir4, disk4) = read_multiple_test_disk(bucket, &[]).await;
+        disk3
+            .delete_volume(bucket, true)
+            .await
+            .expect("third disk bucket should be absent");
+        disk4
+            .delete_volume(bucket, true)
+            .await
+            .expect("fourth disk bucket should be absent");
+        let set = io_primitives_test_set(vec![Some(disk1.clone()), Some(disk2.clone()), Some(disk3), Some(disk4)], 2).await;
+
+        set.delete_prefix(bucket, "prefix")
+            .await
+            .expect("successful deletes and confirmed absence should jointly meet quorum");
+
+        for disk in [disk1, disk2] {
+            assert!(matches!(disk.read_all(bucket, "prefix/object.txt").await, Err(DiskError::FileNotFound)));
+        }
+    }
+
+    #[test]
+    fn delete_prefix_result_reduction_preserves_existing_volume_evidence() {
+        assert_eq!(
+            SetDisks::reduce_delete_prefix_results(
+                vec![
+                    Err(DiskError::FileNotFound),
+                    Err(DiskError::FileNotFound),
+                    Err(DiskError::FileNotFound),
+                    Err(DiskError::DiskNotFound),
+                ],
+                3,
+            ),
+            Ok(())
+        );
+        assert_eq!(
+            SetDisks::reduce_delete_prefix_results(
+                vec![
+                    Err(DiskError::FileNotFound),
+                    Err(DiskError::VolumeNotFound),
+                    Err(DiskError::VolumeNotFound),
+                    Err(DiskError::VolumeNotFound),
+                ],
+                3,
+            ),
+            Ok(())
+        );
+        assert_eq!(
+            SetDisks::reduce_delete_prefix_results(
+                vec![
+                    Ok(()),
+                    Err(DiskError::VolumeNotFound),
+                    Err(DiskError::VolumeNotFound),
+                    Err(DiskError::VolumeNotFound),
+                ],
+                3,
+            ),
+            Ok(())
+        );
+        assert_eq!(
+            SetDisks::reduce_delete_prefix_results(
+                vec![
+                    Err(DiskError::VolumeNotFound),
+                    Err(DiskError::VolumeNotFound),
+                    Err(DiskError::VolumeNotFound),
+                    Err(DiskError::VolumeNotFound),
+                ],
+                3,
+            ),
+            Err(DiskError::VolumeNotFound)
+        );
+        assert_eq!(
+            SetDisks::reduce_delete_prefix_results(
+                vec![Ok(()), Ok(()), Err(DiskError::DiskNotFound), Err(DiskError::DiskNotFound)],
+                3,
+            ),
+            Err(DiskError::ErasureWriteQuorum)
+        );
+        assert_eq!(
+            SetDisks::reduce_delete_prefix_results(
+                vec![
+                    Ok(()),
+                    Err(DiskError::FileAccessDenied),
+                    Err(DiskError::FileAccessDenied),
+                    Err(DiskError::FileAccessDenied),
+                ],
+                3,
+            ),
+            Err(DiskError::FileAccessDenied)
+        );
+    }
+
+    #[tokio::test]
+    async fn delete_prefix_fails_at_quorum_minus_one() {
+        let bucket = "delete-prefix-quorum-minus-one";
+        let (_dir1, disk1) = read_multiple_test_disk(bucket, &[("prefix/object.txt", b"one".as_slice())]).await;
+        let (_dir2, disk2) = read_multiple_test_disk(bucket, &[("prefix/object.txt", b"two".as_slice())]).await;
+        let set = io_primitives_test_set(vec![Some(disk1.clone()), Some(disk2.clone()), None, None], 2).await;
+
+        let err = set
+            .delete_prefix(bucket, "prefix")
+            .await
+            .expect_err("two successful disks must not meet a four-disk write quorum");
+
+        assert_eq!(err, DiskError::ErasureWriteQuorum);
+        for disk in [disk1, disk2] {
+            assert!(matches!(disk.read_all(bucket, "prefix/object.txt").await, Err(DiskError::FileNotFound)));
+        }
+    }
+
+    #[tokio::test]
+    async fn delete_prefix_fails_when_all_disk_slots_are_missing() {
+        let set = io_primitives_test_set(vec![None, None, None, None], 2).await;
+
+        let err = set
+            .delete_prefix("delete-prefix-offline", "prefix")
+            .await
+            .expect_err("an entirely offline set must not report a successful deletion");
+
+        assert_eq!(err, DiskError::ErasureWriteQuorum);
     }
 
     #[tokio::test]

crates/ecstore/src/set_disk/mod.rs

@@ -45,7 +45,10 @@
 
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
 use crate::bucket::metadata_sys;
-use crate::bucket::object_lock::objectlock_sys::{check_object_lock_for_deletion, check_retention_for_modification};
+use crate::bucket::metadata_sys::ObjectLockConfigState;
+use crate::bucket::object_lock::objectlock_sys::{
+    check_object_lock_for_deletion_with_config, check_object_lock_for_deletion_with_state, check_retention_for_modification,
+};
 use crate::bucket::replication::{
     ReplicateDecision, ReplicationObjectBridge, ReplicationState, ReplicationStatusType, VersionPurgeStatusType,
     replication_state_to_filemeta,
@@ -107,7 +110,7 @@ use crate::{
         SnapshotLeaseToken, UpdateMetadataOpts, endpoint::Endpoint, error::DiskError, format::FormatV3, new_disk,
     },
     error::{StorageError, to_object_err},
-    object_api::{GetObjectReader, ObjectInfo, PutObjReader},
+    object_api::{GetObjectReader, NamespaceLockFence, ObjectInfo, ObjectLockConfigSnapshot, PutObjReader},
     // event::name::EventName,
     services::event_notification::{EventArgs, send_event},
     store::init_format::{
@@ -155,9 +158,9 @@ use rustfs_utils::http::headers::{
     CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _,
 };
 use rustfs_utils::http::{
-    SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
-    SUFFIX_RESTORE_OPERATION_ID, contains_key_str, get_header_map, get_str, insert_str, is_object_encryption_marker,
-    remove_header_map,
+    SUFFIX_ACTUAL_OBJECT_SIZE_CAP, SUFFIX_ACTUAL_SIZE, SUFFIX_BUCKET_INCARNATION_ID, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE,
+    SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_RESTORE_OPERATION_ID, contains_key_str, get_header_map, get_str, insert_str,
+    is_object_encryption_marker, remove_header_map,
 };
 use rustfs_utils::{
     HashAlgorithm,
@@ -669,6 +672,7 @@ fn release_materialized_read_lock(bucket: &str, object: &str, read_lock_guard: O
 pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap<String, String>) {
     metadata.remove(RUSTFS_MULTIPART_BUCKET_KEY);
     metadata.remove(RUSTFS_MULTIPART_OBJECT_KEY);
+    rustfs_utils::http::metadata_compat::remove_str(metadata, SUFFIX_BUCKET_INCARNATION_ID);
 }
 
 fn should_persist_encryption_original_size(metadata: &HashMap<String, String>) -> bool {
@@ -956,6 +960,8 @@ pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCl
 pub(crate) use ops::object::body_cache_plaintext_len;
 #[cfg(test)]
 pub(crate) use ops::object::cleanup_rejected_transition_upload_durably;
+#[cfg(test)]
+pub(crate) use ops::object::{PutObjectCommitBarrier, PutObjectCommitPause};
 mod read;
 mod replication;
 pub(crate) mod shard_source;
@@ -3771,7 +3777,49 @@ pub(crate) fn object_lock_delete_check_required(bucket_meta: Option<&crate::buck
     bucket_meta.is_none_or(|meta| meta.object_locking())
 }
 
-async fn check_object_lock_delete(bucket: &str, object: &str, obj_info: &ObjectInfo, opts: &ObjectOptions) -> Result<()> {
+fn restore_expiry_snapshot_matches(obj_info: &ObjectInfo, opts: &ObjectOptions) -> bool {
+    let expected = &opts.transition;
+    expected.expire_restored
+        && expected.status == TRANSITION_COMPLETE
+        && obj_info.transitioned_object.status == TRANSITION_COMPLETE
+        && !obj_info.transitioned_object.name.is_empty()
+        && !obj_info.transitioned_object.tier.is_empty()
+        && expected.tier == obj_info.transitioned_object.tier
+        && expected.expected_remote_name == obj_info.transitioned_object.name
+        && expected.expected_remote_version_id == obj_info.transitioned_object.version_id
+        && !expected.etag.is_empty()
+        && obj_info.etag.as_deref() == Some(expected.etag.as_str())
+        && expected.expected_data_dir.is_some()
+        && expected.expected_data_dir == obj_info.data_dir
+        && obj_info.restore_expires == Some(expected.restore_expiry)
+        && !obj_info.restore_ongoing
+        // Deliberately no `restore_expiry <= now` clause. Whether the restored
+        // copy is due to expire is the ILM evaluator's decision, already made
+        // when it emitted DeleteRestoredAction; re-deriving it here only adds a
+        // way for a legitimate action to be rejected. The stale-event risk it
+        // looks like it covers is already covered above: a re-restore rewrites
+        // `restore_expires`, so a replayed event fails the equality check.
+        && match obj_info.version_id {
+            Some(version_id) => opts.version_id.as_deref().and_then(|value| Uuid::parse_str(value).ok()) == Some(version_id),
+            None => opts.version_id.is_none(),
+        }
+}
+
+async fn check_object_lock_delete(
+    ctx: &InstanceContext,
+    bucket: &str,
+    object: &str,
+    obj_info: &ObjectInfo,
+    opts: &ObjectOptions,
+) -> Result<()> {
+    if crate::bucket::utils::is_meta_bucketname(bucket) {
+        return Ok(());
+    }
+    if opts.transition.expire_restored {
+        return restore_expiry_snapshot_matches(obj_info, opts)
+            .then_some(())
+            .ok_or(StorageError::PreconditionFailed);
+    }
     if set_disk_delete_creates_delete_marker(opts) {
         return Ok(());
     }
@@ -3780,16 +3828,48 @@ async fn check_object_lock_delete(bucket: &str, object: &str, obj_info: &ObjectI
         .object_lock_delete
         .as_ref()
         .is_some_and(|delete_opts| delete_opts.bypass_governance);
-    if check_object_lock_for_deletion(bucket, obj_info, bypass_governance)
-        .await
-        .is_some()
-    {
+    let blocked = match opts.object_lock_config_snapshot.as_deref() {
+        Some(snapshot) => check_object_lock_for_deletion_with_state(snapshot.state(), obj_info, bypass_governance)?.is_some(),
+        None => {
+            let state = metadata_sys::get_object_lock_config_state_in(ctx, bucket).await?;
+            check_object_lock_for_deletion_with_state(&state, obj_info, bypass_governance)?.is_some()
+        }
+    };
+    if blocked {
         return Err(StorageError::PrefixAccessDenied(bucket.to_string(), object.to_string()));
     }
 
     Ok(())
 }
 
+fn ensure_delete_commit_locks_held(
+    lock_guard: Option<&ObjectLockDiagGuard>,
+    bucket: &str,
+    object: &str,
+    opts: &ObjectOptions,
+) -> Result<()> {
+    if lock_guard.is_some_and(ObjectLockDiagGuard::is_lock_lost)
+        || opts
+            .namespace_lock_fence
+            .as_ref()
+            .is_some_and(NamespaceLockFence::is_lock_lost)
+        || opts
+            .bucket_lifecycle_lock_fence
+            .as_ref()
+            .is_some_and(NamespaceLockFence::is_lock_lost)
+    {
+        return Err(StorageError::NamespaceLockQuorumUnavailable {
+            mode: "delete_object_commit",
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            required: 1,
+            achieved: 0,
+        });
+    }
+
+    Ok(())
+}
+
 fn set_disk_delete_creates_delete_marker(opts: &ObjectOptions) -> bool {
     opts.version_id.is_none() && opts.versioned && !opts.version_suspended
 }
@@ -5672,7 +5752,7 @@ mod tests {
             .await
             .expect("outer write lock should be acquired");
 
-        timeout(
+        let result = timeout(
             Duration::from_secs(1),
             set_disks.delete_object(
                 "bucket",
@@ -5684,8 +5764,14 @@ mod tests {
             ),
         )
         .await
-        .expect("broad prefix delete must not wait on a literal prefix namespace lock")
-        .expect("empty test disks should allow broad prefix cleanup");
+        .expect("broad prefix delete must not wait on a literal prefix namespace lock");
+
+        if let Err(err) = result {
+            assert!(
+                !err.to_string().to_ascii_lowercase().contains("lock"),
+                "broad prefix delete returned a lock error: {err}"
+            );
+        }
     }
 
     #[tokio::test(flavor = "multi_thread")]
@@ -5705,7 +5791,7 @@ mod tests {
             .await
             .expect("outer write lock should be acquired");
 
-        timeout(
+        let result = timeout(
             Duration::from_secs(1),
             set_disks.delete_object(
                 "bucket",
@@ -5719,8 +5805,14 @@ mod tests {
             ),
         )
         .await
-        .expect("no_lock exact prefix delete path must not wait for the outer lock")
-        .expect("empty test disks should allow exact prefix cleanup");
+        .expect("no_lock exact prefix delete path must not wait for the outer lock");
+
+        if let Err(err) = result {
+            assert!(
+                !err.to_string().to_ascii_lowercase().contains("lock"),
+                "no_lock exact prefix delete returned a lock error: {err}"
+            );
+        }
     }
 
     #[tokio::test(flavor = "multi_thread")]
@@ -8556,16 +8648,105 @@ mod tests {
         let opts = ObjectOptions {
             version_id: Some(Uuid::new_v4().to_string()),
             versioned: true,
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
             ..Default::default()
         };
 
-        let err = check_object_lock_delete("bucket", "object", &obj_info, &opts)
+        let err = check_object_lock_delete(&bootstrap_ctx(), "bucket", "object", &obj_info, &opts)
             .await
             .expect_err("COMPLIANCE retention must block explicit version deletion");
 
         assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)));
     }
 
+    #[tokio::test]
+    async fn test_check_object_lock_delete_allows_retained_restored_copy_expiry() {
+        let retain_until = OffsetDateTime::now_utc() + Duration::from_secs(60 * 60 * 24 * 60);
+        let restore_expiry = OffsetDateTime::now_utc() - Duration::from_secs(1);
+        let version_id = Uuid::new_v4();
+        let data_dir = Uuid::new_v4();
+        let obj_info = ObjectInfo {
+            version_id: Some(version_id),
+            data_dir: Some(data_dir),
+            etag: Some("etag".to_string()),
+            transitioned_object: TransitionedObject {
+                name: "remote-object".to_string(),
+                tier: "tier".to_string(),
+                status: TRANSITION_COMPLETE.to_string(),
+                ..Default::default()
+            },
+            restore_expires: Some(restore_expiry),
+            user_defined: Arc::new(HashMap::from([
+                (
+                    X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(),
+                    s3s::dto::ObjectLockRetentionMode::COMPLIANCE.to_string(),
+                ),
+                (
+                    X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(),
+                    retain_until.format(&time::format_description::well_known::Rfc3339).unwrap(),
+                ),
+            ])),
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            version_id: Some(version_id.to_string()),
+            versioned: true,
+            transition: crate::bucket::lifecycle::lifecycle::TransitionOptions {
+                status: TRANSITION_COMPLETE.to_string(),
+                tier: "tier".to_string(),
+                etag: "etag".to_string(),
+                expected_data_dir: Some(data_dir),
+                expected_remote_name: "remote-object".to_string(),
+                restore_expiry,
+                expire_restored: true,
+                ..Default::default()
+            },
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
+            ..Default::default()
+        };
+
+        check_object_lock_delete(&bootstrap_ctx(), "bucket", "object", &obj_info, &opts)
+            .await
+            .expect("restore expiry only strips the local copy and must preserve the retained logical version");
+    }
+
+    #[tokio::test]
+    async fn test_check_object_lock_delete_rejects_stale_restored_copy_expiry() {
+        let restore_expiry = OffsetDateTime::now_utc() - Duration::from_secs(1);
+        let data_dir = Uuid::new_v4();
+        let obj_info = ObjectInfo {
+            data_dir: Some(data_dir),
+            etag: Some("etag".to_string()),
+            transitioned_object: TransitionedObject {
+                name: "remote-object".to_string(),
+                tier: "tier".to_string(),
+                status: TRANSITION_COMPLETE.to_string(),
+                ..Default::default()
+            },
+            restore_expires: Some(restore_expiry + Duration::from_secs(60)),
+            ..Default::default()
+        };
+        let opts = ObjectOptions {
+            versioned: true,
+            transition: crate::bucket::lifecycle::lifecycle::TransitionOptions {
+                status: TRANSITION_COMPLETE.to_string(),
+                tier: "tier".to_string(),
+                etag: "etag".to_string(),
+                expected_data_dir: Some(data_dir),
+                expected_remote_name: "remote-object".to_string(),
+                restore_expiry,
+                expire_restored: true,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+
+        let err = check_object_lock_delete(&bootstrap_ctx(), "bucket", "object", &obj_info, &opts)
+            .await
+            .expect_err("a renewed restored copy must reject the stale expiry task");
+        assert!(matches!(err, StorageError::PreconditionFailed));
+    }
+
     #[tokio::test]
     async fn test_check_object_lock_delete_allows_versioned_delete_marker_creation() {
         let retain_until = OffsetDateTime::now_utc() + Duration::from_secs(60 * 60 * 24 * 60);
@@ -8590,7 +8771,7 @@ mod tests {
             ..Default::default()
         };
 
-        check_object_lock_delete("bucket", "object", &obj_info, &opts)
+        check_object_lock_delete(&bootstrap_ctx(), "bucket", "object", &obj_info, &opts)
             .await
             .expect("versioned delete marker creation should not delete the locked version");
     }
@@ -9881,6 +10062,7 @@ mod tests {
         let payload = (0..(BLOCK_SIZE_V2 + 17)).map(|idx| (idx % 251) as u8).collect::<Vec<_>>();
         let opts = ObjectOptions {
             no_lock: true,
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
             ..Default::default()
         };
 
@@ -10006,7 +10188,12 @@ mod tests {
             let bucket = "snapshot-streaming-delete";
             let object = "object";
             let body = vec![0x41; 2 * 1024 * 1024];
-            let opts = ObjectOptions::default();
+            let opts = ObjectOptions {
+                object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
+                    ObjectLockConfigState::ConfirmedAbsent,
+                ))),
+                ..Default::default()
+            };
 
             set_disks
                 .make_bucket(bucket, &MakeBucketOptions::default())
@@ -10058,7 +10245,12 @@ mod tests {
             let bucket = "snapshot-streaming-delete-objects";
             let object = "object";
             let body = vec![0x41; 2 * 1024 * 1024];
-            let opts = ObjectOptions::default();
+            let opts = ObjectOptions {
+                object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
+                    ObjectLockConfigState::ConfirmedAbsent,
+                ))),
+                ..Default::default()
+            };
 
             set_disks
                 .make_bucket(bucket, &MakeBucketOptions::default())
@@ -10919,6 +11111,71 @@ mod tests {
         assert_eq!(restored, payload);
     }
 
+    /// The other half of the suspended-versioning delete contract: a client
+    /// that drains such a bucket lists the null delete marker and then purges
+    /// it as `?versionId=null`, which is what `nuke_bucket` does before
+    /// `DeleteBucket`. That purge must succeed — if it is rejected the marker's
+    /// `xl.meta` survives, and `DeleteBucket`'s raw disk scan then reports
+    /// `BucketNotEmpty` for a bucket the client has already emptied.
+    #[tokio::test]
+    async fn set_level_explicit_null_version_delete_purges_the_null_delete_marker() {
+        let set_disks = make_local_bucket_test_set_disks().await;
+        let bucket = "bucket-null-marker-purge";
+        let object = "object.txt";
+        // The delete path reads versioned/suspended from the bucket-config
+        // snapshot, not from `opts`, so inject a real Suspended config —
+        // otherwise `from_file_info` never synthesizes the null version id and
+        // the branch under test is not reached.
+        let suspended = crate::bucket::replication::DeleteReplicationConfigSnapshot::from_configs_for_test(
+            s3s::dto::VersioningConfiguration {
+                status: Some(s3s::dto::BucketVersioningStatus::from_static(s3s::dto::BucketVersioningStatus::SUSPENDED)),
+                ..Default::default()
+            },
+            None,
+        );
+        let opts = ObjectOptions {
+            no_lock: true,
+            version_suspended: true,
+            delete_replication_config_snapshot: Some(Arc::new(suspended)),
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
+            ..Default::default()
+        };
+
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        let mut reader = PutObjReader::from_vec(b"suspended version body".to_vec());
+        set_disks
+            .put_object(bucket, object, &mut reader, &opts)
+            .await
+            .expect("suspended-version object should be written");
+
+        let marker = set_disks
+            .delete_object(bucket, object, opts.clone())
+            .await
+            .expect("version-suspended delete should create a null marker");
+        assert!(marker.delete_marker);
+        assert_eq!(marker.version_id, Some(Uuid::nil()));
+
+        let (_deleted, errs) = set_disks
+            .delete_objects(
+                bucket,
+                vec![ObjectToDelete {
+                    object_name: object.to_string(),
+                    version_id: Some(Uuid::nil()),
+                    ..Default::default()
+                }],
+                opts.clone(),
+            )
+            .await;
+
+        assert!(
+            errs.iter().all(Option::is_none),
+            "explicit null-version purge of the null delete marker must succeed, got {errs:?}"
+        );
+    }
+
     #[tokio::test]
     async fn set_level_version_suspended_delete_creates_null_delete_marker() {
         let set_disks = make_local_bucket_test_set_disks().await;
@@ -10927,6 +11184,7 @@ mod tests {
         let opts = ObjectOptions {
             no_lock: true,
             version_suspended: true,
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
             ..Default::default()
         };
 
@@ -11016,6 +11274,9 @@ mod tests {
             let missing = "missing.txt";
             let opts = ObjectOptions {
                 no_lock: true,
+                object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
+                    ObjectLockConfigState::ConfirmedAbsent,
+                ))),
                 ..Default::default()
             };
 

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

@@ -162,6 +162,86 @@ fn map_upload_id_metadata_error(bucket: &str, object: &str, upload_id: &str, err
     err.into()
 }
 
+fn multipart_bucket_incarnation_id(metadata: &HashMap<String, String>) -> Result<Option<Uuid>> {
+    let Some(value) = rustfs_utils::http::metadata_compat::get_consistent_str(metadata, SUFFIX_BUCKET_INCARNATION_ID) else {
+        if rustfs_utils::http::metadata_compat::contains_key_str(metadata, SUFFIX_BUCKET_INCARNATION_ID) {
+            return Err(Error::other("invalid multipart bucket incarnation metadata"));
+        }
+        return Ok(None);
+    };
+    let incarnation = Uuid::parse_str(value).map_err(|_| Error::other("invalid multipart bucket incarnation metadata"))?;
+    if incarnation.is_nil() {
+        return Err(Error::other("invalid multipart bucket incarnation metadata"));
+    }
+    Ok(Some(incarnation))
+}
+
+fn multipart_bucket_incarnation_matches(metadata: &HashMap<String, String>, expected: Uuid) -> bool {
+    matches!(multipart_bucket_incarnation_id(metadata), Ok(Some(actual)) if actual == expected)
+}
+
+fn validate_multipart_bucket_incarnation(
+    metadata: &HashMap<String, String>,
+    bucket: &str,
+    object: &str,
+    upload_id: &str,
+    expected: Option<Uuid>,
+    upload_initiated: Option<OffsetDateTime>,
+    bucket_created: Option<OffsetDateTime>,
+) -> Result<()> {
+    let Some(expected) = expected else {
+        return Ok(());
+    };
+    match multipart_bucket_incarnation_id(metadata) {
+        Ok(Some(actual)) if actual == expected => return Ok(()),
+        Ok(None)
+            if matches!(
+                (upload_initiated, bucket_created),
+                (Some(upload_initiated), Some(bucket_created)) if upload_initiated >= bucket_created
+            ) =>
+        {
+            return Ok(());
+        }
+        _ => {}
+    }
+    Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
+}
+
+async fn ensure_multipart_bucket_incarnation(
+    ctx: &crate::runtime::instance::InstanceContext,
+    fi: &FileInfo,
+    bucket: &str,
+    object: &str,
+    upload_id: &str,
+    expected: Option<Uuid>,
+) -> Result<()> {
+    let bucket_created = if expected.is_some() && matches!(multipart_bucket_incarnation_id(&fi.metadata), Ok(None)) {
+        Some(metadata_sys::created_at_in(ctx, bucket).await?)
+    } else {
+        None
+    };
+    validate_multipart_bucket_incarnation(&fi.metadata, bucket, object, upload_id, expected, fi.mod_time, bucket_created)
+}
+
+fn ensure_multipart_bucket_lifecycle_lock_held(bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> {
+    let Some(fence) = opts.bucket_lifecycle_lock_fence.as_ref() else {
+        if opts.expected_bucket_incarnation_id.is_some() && !crate::bucket::utils::is_meta_bucketname(bucket) {
+            return Err(Error::other("multipart bucket lifecycle lock fence is missing"));
+        }
+        return Ok(());
+    };
+    if fence.is_lock_lost() {
+        return Err(StorageError::NamespaceLockQuorumUnavailable {
+            mode: "multipart_bucket_generation",
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            required: 1,
+            achieved: 0,
+        });
+    }
+    Ok(())
+}
+
 fn empty_upload_fallback_possible(successful_responses: usize, errs: &[Option<DiskError>]) -> bool {
     successful_responses == 0
         && errs.iter().any(|err| matches!(err, Some(DiskError::FileNotFound)))
@@ -451,6 +531,200 @@ impl SetDisks {
 
         Ok((fi, parts_metadata))
     }
+
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) async fn list_multipart_uploads_for_incarnation(
+        &self,
+        bucket: &str,
+        prefix: &str,
+        key_marker: Option<String>,
+        upload_id_marker: Option<String>,
+        delimiter: Option<String>,
+        max_uploads: usize,
+        expected_incarnation_id: Option<Uuid>,
+    ) -> Result<ListMultipartsInfo> {
+        let disks = self.disks.read().await.clone();
+        if disks.is_empty() {
+            return Err(Error::ErasureReadQuorum);
+        }
+        let discovery_quorum = if self.default_parity_count == 0 {
+            disks.len()
+        } else {
+            (disks.len() / 2).max(1)
+        };
+        let mut discovery_errors = (0..disks.len()).map(|_| Some(DiskError::DiskNotFound)).collect::<Vec<_>>();
+        let mut candidate_counts = HashMap::<String, usize>::new();
+        let mut discovery_tasks = JoinSet::new();
+        for (index, disk) in disks.iter().enumerate() {
+            let disk = disk.clone();
+            let bucket = bucket.to_string();
+            discovery_tasks.spawn(async move {
+                let result = match disk {
+                    Some(disk) => multipart_upload_paths_on_disk(disk, &bucket).await,
+                    None => Err(DiskError::DiskNotFound),
+                };
+                (index, result)
+            });
+        }
+
+        while let Some(task_result) = discovery_tasks.join_next().await {
+            let Ok((index, result)) = task_result else {
+                continue;
+            };
+            match result {
+                Ok(paths) => {
+                    discovery_errors[index] = None;
+                    for path in paths {
+                        *candidate_counts.entry(path).or_insert(0) += 1;
+                    }
+                }
+                Err(err) => discovery_errors[index] = Some(err),
+            }
+        }
+
+        if let Some(err) = reduce_read_quorum_errs(&discovery_errors, OBJECT_OP_IGNORED_ERRS, discovery_quorum) {
+            return Err(to_object_err(err.into(), vec![bucket, prefix]));
+        }
+
+        let candidate_paths = candidate_counts
+            .into_iter()
+            .filter_map(|(path, count)| (count >= discovery_quorum).then_some(path))
+            .collect::<Vec<_>>();
+        let listed_uploads = stream::iter(candidate_paths)
+            .map(|upload_path| {
+                let disks = &disks;
+                async move {
+                    let (sha_dir, raw_upload_id) = upload_path
+                        .rsplit_once('/')
+                        .filter(|(sha_dir, upload_id)| !sha_dir.is_empty() && !upload_id.is_empty())
+                        .ok_or(DiskError::CorruptedFormat)?;
+                    let (parts_metadata, errs) = Self::read_all_fileinfo(
+                        disks,
+                        bucket,
+                        RUSTFS_META_MULTIPART_BUCKET,
+                        &upload_path,
+                        "",
+                        false,
+                        false,
+                        false,
+                    )
+                    .await?;
+                    let missing_metadata = errs
+                        .iter()
+                        .filter(|err| matches!(err, Some(DiskError::FileNotFound | DiskError::VolumeNotFound)))
+                        .count();
+                    if missing_metadata >= discovery_quorum {
+                        if expected_incarnation_id.is_some() {
+                            return Ok(None);
+                        }
+                        // Completion moves the authoritative upload metadata into the
+                        // committed object before it removes the staging directory. A
+                        // crash in that window intentionally leaves a reclaimable
+                        // upload directory whose object name can still be proven for
+                        // an exact-key listing by matching the namespace hash.
+                        if !prefix.is_empty() && sha_dir == Self::get_multipart_sha_dir(bucket, prefix) {
+                            let initiated = raw_upload_id
+                                .rsplit_once('x')
+                                .and_then(|(_, timestamp)| timestamp.parse::<i128>().ok())
+                                .and_then(|timestamp| OffsetDateTime::from_unix_timestamp_nanos(timestamp).ok());
+                            return Ok(Some(MultipartInfo {
+                                bucket: bucket.to_owned(),
+                                object: prefix.to_owned(),
+                                upload_id: runtime_sources::deployment_upload_id(raw_upload_id),
+                                initiated,
+                                ..Default::default()
+                            }));
+                        }
+                        return Ok(None);
+                    }
+                    let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?;
+                    let read_quorum = usize::try_from(read_quorum).map_err(|_| DiskError::ErasureReadQuorum)?;
+                    if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) {
+                        return Err(err);
+                    }
+                    let (_, mod_time, etag) = Self::list_online_disks(disks, &parts_metadata, &errs, read_quorum);
+                    let file_info = Self::pick_valid_fileinfo(&parts_metadata, mod_time, etag, read_quorum)?;
+                    if expected_incarnation_id
+                        .is_some_and(|expected| !multipart_bucket_incarnation_matches(&file_info.metadata, expected))
+                    {
+                        return Ok(None);
+                    }
+
+                    let object = match (
+                        file_info.metadata.get(RUSTFS_MULTIPART_BUCKET_KEY),
+                        file_info.metadata.get(RUSTFS_MULTIPART_OBJECT_KEY),
+                    ) {
+                        (Some(stored_bucket), Some(object)) if stored_bucket == bucket && !object.is_empty() => object.clone(),
+                        _ => return Err(DiskError::CorruptedFormat),
+                    };
+                    if !object.starts_with(prefix) {
+                        return Ok(None);
+                    }
+
+                    let initiated = raw_upload_id
+                        .rsplit_once('x')
+                        .and_then(|(_, timestamp)| timestamp.parse::<i128>().ok())
+                        .and_then(|timestamp| OffsetDateTime::from_unix_timestamp_nanos(timestamp).ok())
+                        .or(file_info.mod_time);
+
+                    Ok(Some(MultipartInfo {
+                        bucket: bucket.to_owned(),
+                        object,
+                        upload_id: runtime_sources::deployment_upload_id(raw_upload_id),
+                        initiated,
+                        ..Default::default()
+                    }))
+                }
+            })
+            .buffer_unordered(MULTIPART_LIST_IO_CONCURRENCY)
+            .collect::<Vec<disk::error::Result<Option<MultipartInfo>>>>()
+            .await;
+
+        let mut uploads = Vec::with_capacity(listed_uploads.len());
+        for result in listed_uploads {
+            if let Some(upload) = result.map_err(Error::from)? {
+                uploads.push(upload);
+            }
+        }
+
+        let mut common_prefixes = HashSet::new();
+        let mut unfolded_uploads = Vec::with_capacity(uploads.len());
+        let delimiter_value = delimiter.as_deref().filter(|delimiter| !delimiter.is_empty());
+        for upload in uploads {
+            let Some(delimiter) = delimiter_value else {
+                unfolded_uploads.push(upload);
+                continue;
+            };
+            let suffix = upload.object.strip_prefix(prefix).ok_or(DiskError::CorruptedFormat)?;
+            if let Some((common_prefix, _)) = suffix.split_once(delimiter) {
+                common_prefixes.insert(format!("{prefix}{common_prefix}{delimiter}"));
+            } else {
+                unfolded_uploads.push(upload);
+            }
+        }
+
+        let page = paginate_multipart_listing(
+            unfolded_uploads,
+            common_prefixes.into_iter().collect(),
+            key_marker.as_deref(),
+            key_marker.as_ref().and(upload_id_marker.as_deref()),
+            max_uploads,
+            false,
+        );
+
+        Ok(ListMultipartsInfo {
+            key_marker: key_marker.to_owned(),
+            upload_id_marker: upload_id_marker.to_owned(),
+            next_key_marker: page.next_key_marker,
+            next_upload_id_marker: page.next_upload_id_marker,
+            max_uploads,
+            is_truncated: page.is_truncated,
+            uploads: page.uploads,
+            common_prefixes: page.common_prefixes,
+            prefix: prefix.to_owned(),
+            delimiter: delimiter.to_owned(),
+        })
+    }
 }
 
 #[async_trait::async_trait]
@@ -498,6 +772,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
 
         let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
+        ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
+            .await?;
 
         let write_quorum = fi.write_quorum(self.default_write_quorum());
 
@@ -747,7 +1023,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                         .await?,
                 )
             };
-            self.check_upload_id_exists(bucket, object, upload_id, false).await?;
+            let (commit_fi, _) = self.check_upload_id_exists(bucket, object, upload_id, false).await?;
+            ensure_multipart_bucket_incarnation(
+                &self.ctx,
+                &commit_fi,
+                bucket,
+                object,
+                upload_id,
+                opts.expected_bucket_incarnation_id,
+            )
+            .await?;
             #[cfg(test)]
             pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await;
             if _upload_commit_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
@@ -759,6 +1044,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     achieved: 0,
                 });
             }
+            ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
             let _ = self
                 .rename_part(
@@ -820,6 +1106,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             .acquire_multipart_upload_read_lock("list_object_parts", bucket, object, upload_id, opts)
             .await?;
         let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, false).await?;
+        ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
+            .await?;
 
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
 
@@ -927,6 +1215,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             ret.next_part_number_marker = ret.parts.last().map(|v| v.part_num).unwrap_or_default();
         }
 
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
         Ok(ret)
     }
 
@@ -940,179 +1229,44 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         delimiter: Option<String>,
         max_uploads: usize,
     ) -> Result<ListMultipartsInfo> {
-        let disks = self.disks.read().await.clone();
-        if disks.is_empty() {
-            return Err(Error::ErasureReadQuorum);
-        }
-        let discovery_quorum = if self.default_parity_count == 0 {
-            disks.len()
+        let bucket_lifecycle_guard = if crate::bucket::utils::is_meta_bucketname(bucket) {
+            None
         } else {
-            (disks.len() / 2).max(1)
+            Some(
+                metadata_sys::object_store_in(&self.ctx)
+                    .await?
+                    .acquire_bucket_lifecycle_read_lock(bucket)
+                    .await?,
+            )
         };
-        let mut discovery_errors = (0..disks.len()).map(|_| Some(DiskError::DiskNotFound)).collect::<Vec<_>>();
-        let mut candidate_counts = HashMap::<String, usize>::new();
-        let mut discovery_tasks = JoinSet::new();
-        for (index, disk) in disks.iter().enumerate() {
-            let disk = disk.clone();
-            let bucket = bucket.to_string();
-            discovery_tasks.spawn(async move {
-                let result = match disk {
-                    Some(disk) => multipart_upload_paths_on_disk(disk, &bucket).await,
-                    None => Err(DiskError::DiskNotFound),
-                };
-                (index, result)
+        let expected_incarnation_id = if bucket_lifecycle_guard.is_some() {
+            Some(metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await?)
+        } else {
+            None
+        };
+        let result = self
+            .list_multipart_uploads_for_incarnation(
+                bucket,
+                prefix,
+                key_marker,
+                upload_id_marker,
+                delimiter,
+                max_uploads,
+                expected_incarnation_id,
+            )
+            .await;
+
+        if bucket_lifecycle_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+            return Err(StorageError::NamespaceLockQuorumUnavailable {
+                mode: "multipart_bucket_generation",
+                bucket: bucket.to_string(),
+                object: prefix.to_string(),
+                required: 1,
+                achieved: 0,
             });
         }
 
-        while let Some(task_result) = discovery_tasks.join_next().await {
-            let Ok((index, result)) = task_result else {
-                continue;
-            };
-            match result {
-                Ok(paths) => {
-                    discovery_errors[index] = None;
-                    for path in paths {
-                        *candidate_counts.entry(path).or_insert(0) += 1;
-                    }
-                }
-                Err(err) => discovery_errors[index] = Some(err),
-            }
-        }
-
-        if let Some(err) = reduce_read_quorum_errs(&discovery_errors, OBJECT_OP_IGNORED_ERRS, discovery_quorum) {
-            return Err(to_object_err(err.into(), vec![bucket, prefix]));
-        }
-
-        let candidate_paths = candidate_counts
-            .into_iter()
-            .filter_map(|(path, count)| (count >= discovery_quorum).then_some(path))
-            .collect::<Vec<_>>();
-        let listed_uploads = stream::iter(candidate_paths)
-            .map(|upload_path| {
-                let disks = &disks;
-                async move {
-                    let (sha_dir, raw_upload_id) = upload_path
-                        .rsplit_once('/')
-                        .filter(|(sha_dir, upload_id)| !sha_dir.is_empty() && !upload_id.is_empty())
-                        .ok_or(DiskError::CorruptedFormat)?;
-                    let (parts_metadata, errs) = Self::read_all_fileinfo(
-                        disks,
-                        bucket,
-                        RUSTFS_META_MULTIPART_BUCKET,
-                        &upload_path,
-                        "",
-                        false,
-                        false,
-                        false,
-                    )
-                    .await?;
-                    let missing_metadata = errs
-                        .iter()
-                        .filter(|err| matches!(err, Some(DiskError::FileNotFound | DiskError::VolumeNotFound)))
-                        .count();
-                    if missing_metadata >= discovery_quorum {
-                        // Completion moves the authoritative upload metadata into the
-                        // committed object before it removes the staging directory. A
-                        // crash in that window intentionally leaves a reclaimable
-                        // upload directory whose object name can still be proven for
-                        // an exact-key listing by matching the namespace hash.
-                        if !prefix.is_empty() && sha_dir == Self::get_multipart_sha_dir(bucket, prefix) {
-                            let initiated = raw_upload_id
-                                .rsplit_once('x')
-                                .and_then(|(_, timestamp)| timestamp.parse::<i128>().ok())
-                                .and_then(|timestamp| OffsetDateTime::from_unix_timestamp_nanos(timestamp).ok());
-                            return Ok(Some(MultipartInfo {
-                                bucket: bucket.to_owned(),
-                                object: prefix.to_owned(),
-                                upload_id: runtime_sources::deployment_upload_id(raw_upload_id),
-                                initiated,
-                                ..Default::default()
-                            }));
-                        }
-                        return Ok(None);
-                    }
-                    let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?;
-                    let read_quorum = usize::try_from(read_quorum).map_err(|_| DiskError::ErasureReadQuorum)?;
-                    if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) {
-                        return Err(err);
-                    }
-                    let (_, mod_time, etag) = Self::list_online_disks(disks, &parts_metadata, &errs, read_quorum);
-                    let file_info = Self::pick_valid_fileinfo(&parts_metadata, mod_time, etag, read_quorum)?;
-
-                    let object = match (
-                        file_info.metadata.get(RUSTFS_MULTIPART_BUCKET_KEY),
-                        file_info.metadata.get(RUSTFS_MULTIPART_OBJECT_KEY),
-                    ) {
-                        (Some(stored_bucket), Some(object)) if stored_bucket == bucket && !object.is_empty() => object.clone(),
-                        _ => return Err(DiskError::CorruptedFormat),
-                    };
-                    if !object.starts_with(prefix) {
-                        return Ok(None);
-                    }
-
-                    let initiated = raw_upload_id
-                        .rsplit_once('x')
-                        .and_then(|(_, timestamp)| timestamp.parse::<i128>().ok())
-                        .and_then(|timestamp| OffsetDateTime::from_unix_timestamp_nanos(timestamp).ok())
-                        .or(file_info.mod_time);
-
-                    Ok(Some(MultipartInfo {
-                        bucket: bucket.to_owned(),
-                        object,
-                        upload_id: runtime_sources::deployment_upload_id(raw_upload_id),
-                        initiated,
-                        ..Default::default()
-                    }))
-                }
-            })
-            .buffer_unordered(MULTIPART_LIST_IO_CONCURRENCY)
-            .collect::<Vec<disk::error::Result<Option<MultipartInfo>>>>()
-            .await;
-
-        let mut uploads = Vec::with_capacity(listed_uploads.len());
-        for result in listed_uploads {
-            if let Some(upload) = result.map_err(Error::from)? {
-                uploads.push(upload);
-            }
-        }
-
-        let mut common_prefixes = HashSet::new();
-        let mut unfolded_uploads = Vec::with_capacity(uploads.len());
-        let delimiter_value = delimiter.as_deref().filter(|delimiter| !delimiter.is_empty());
-        for upload in uploads {
-            let Some(delimiter) = delimiter_value else {
-                unfolded_uploads.push(upload);
-                continue;
-            };
-            let suffix = upload.object.strip_prefix(prefix).ok_or(DiskError::CorruptedFormat)?;
-            if let Some((common_prefix, _)) = suffix.split_once(delimiter) {
-                common_prefixes.insert(format!("{prefix}{common_prefix}{delimiter}"));
-            } else {
-                unfolded_uploads.push(upload);
-            }
-        }
-
-        let page = paginate_multipart_listing(
-            unfolded_uploads,
-            common_prefixes.into_iter().collect(),
-            key_marker.as_deref(),
-            key_marker.as_ref().and(upload_id_marker.as_deref()),
-            max_uploads,
-            false,
-        );
-
-        Ok(ListMultipartsInfo {
-            key_marker: key_marker.to_owned(),
-            upload_id_marker: upload_id_marker.to_owned(),
-            next_key_marker: page.next_key_marker,
-            next_upload_id_marker: page.next_upload_id_marker,
-            max_uploads,
-            is_truncated: page.is_truncated,
-            uploads: page.uploads,
-            common_prefixes: page.common_prefixes,
-            prefix: prefix.to_owned(),
-            delimiter: delimiter.to_owned(),
-        })
+        result
     }
 
     #[tracing::instrument(skip(self))]
@@ -1225,6 +1379,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         user_defined.insert(RUSTFS_MULTIPART_BUCKET_KEY.to_string(), bucket.to_string());
         user_defined.insert(RUSTFS_MULTIPART_OBJECT_KEY.to_string(), object.to_string());
+        if let Some(incarnation_id) = opts.expected_bucket_incarnation_id {
+            insert_str(&mut user_defined, SUFFIX_BUCKET_INCARNATION_ID, incarnation_id.to_string());
+        }
 
         let (shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi);
         let mod_time = opts.mod_time.unwrap_or_else(OffsetDateTime::now_utc);
@@ -1243,6 +1400,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         let upload_path = Self::get_upload_id_dir(bucket, object, upload_uuid.as_str());
 
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
         Self::write_unique_file_info(
             &shuffle_disks,
             bucket,
@@ -1278,6 +1436,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             .check_upload_id_exists(bucket, object, upload_id, false)
             .await
             .map_err(|e| to_object_err(e, vec![bucket, object, upload_id]))?;
+        ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
+            .await?;
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
         Ok(MultipartInfo {
             bucket: bucket.to_owned(),
@@ -1297,6 +1458,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             .acquire_multipart_upload_write_lock("abort_multipart_upload", bucket, object, upload_id, opts)
             .await?;
         let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
+        ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
+            .await?;
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
 
         self.delete_all_with_quorum(
@@ -1348,6 +1512,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         let expected_restore_operation_id = restore_commit_operation_id_from_metadata(&opts.user_defined)?;
         let (mut fi, files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
+        ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
+            .await?;
         let has_layout_candidate = range_seek_rollout_enabled
             && fi
                 .data_dir
@@ -1786,6 +1952,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 achieved: 0,
             });
         }
+        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
         self.require_current_restore_operation_id(
             bucket,
@@ -1979,6 +2146,107 @@ mod tests {
     use tempfile::TempDir;
     use tokio::sync::{Notify, RwLock};
 
+    #[test]
+    fn multipart_bucket_incarnation_metadata_is_consistent_and_non_nil() {
+        let incarnation = Uuid::new_v4();
+        let mut metadata = HashMap::new();
+        insert_str(&mut metadata, SUFFIX_BUCKET_INCARNATION_ID, incarnation.to_string());
+        assert_eq!(multipart_bucket_incarnation_id(&metadata).unwrap(), Some(incarnation));
+
+        metadata.insert("x-minio-internal-bucket-incarnation-id".to_string(), Uuid::new_v4().to_string());
+        assert!(multipart_bucket_incarnation_id(&metadata).is_err());
+
+        let mut nil_metadata = HashMap::new();
+        insert_str(&mut nil_metadata, SUFFIX_BUCKET_INCARNATION_ID, Uuid::nil().to_string());
+        assert!(multipart_bucket_incarnation_id(&nil_metadata).is_err());
+    }
+
+    #[test]
+    fn multipart_bucket_incarnation_gate_accepts_only_current_or_same_lifetime_legacy_uploads() {
+        let expected = Uuid::new_v4();
+        let stale = Uuid::new_v4();
+        let bucket_created = OffsetDateTime::now_utc();
+        let upload_initiated = bucket_created + time::Duration::seconds(1);
+        let mut current_metadata = HashMap::new();
+        insert_str(&mut current_metadata, SUFFIX_BUCKET_INCARNATION_ID, expected.to_string());
+        assert!(multipart_bucket_incarnation_matches(&current_metadata, expected));
+        validate_multipart_bucket_incarnation(&current_metadata, "bucket", "object", "upload", Some(expected), None, None)
+            .expect("a matching stamped upload should pass");
+
+        let missing_metadata = HashMap::new();
+        assert!(!multipart_bucket_incarnation_matches(&missing_metadata, expected));
+        validate_multipart_bucket_incarnation(
+            &missing_metadata,
+            "bucket",
+            "object",
+            "upload",
+            Some(expected),
+            Some(upload_initiated),
+            Some(bucket_created),
+        )
+        .expect("a legacy upload initiated during the current bucket lifetime should pass");
+        assert!(matches!(
+            validate_multipart_bucket_incarnation(
+                &missing_metadata,
+                "bucket",
+                "object",
+                "upload",
+                Some(expected),
+                Some(bucket_created - time::Duration::seconds(1)),
+                Some(bucket_created),
+            ),
+            Err(StorageError::InvalidUploadID(..))
+        ));
+        assert!(matches!(
+            validate_multipart_bucket_incarnation(
+                &missing_metadata,
+                "bucket",
+                "object",
+                "upload",
+                Some(expected),
+                None,
+                Some(bucket_created),
+            ),
+            Err(StorageError::InvalidUploadID(..))
+        ));
+
+        let mut stale_metadata = HashMap::new();
+        insert_str(&mut stale_metadata, SUFFIX_BUCKET_INCARNATION_ID, stale.to_string());
+        assert!(!multipart_bucket_incarnation_matches(&stale_metadata, expected));
+        assert!(matches!(
+            validate_multipart_bucket_incarnation(
+                &stale_metadata,
+                "bucket",
+                "object",
+                "upload",
+                Some(expected),
+                Some(upload_initiated),
+                Some(bucket_created),
+            ),
+            Err(StorageError::InvalidUploadID(..))
+        ));
+    }
+
+    #[test]
+    fn multipart_commit_rejects_missing_or_lost_bucket_lifecycle_fence() {
+        let expected = Uuid::new_v4();
+        let missing = ObjectOptions {
+            expected_bucket_incarnation_id: Some(expected),
+            ..Default::default()
+        };
+        assert!(ensure_multipart_bucket_lifecycle_lock_held("bucket", "object", &missing).is_err());
+
+        let lost = ObjectOptions {
+            expected_bucket_incarnation_id: Some(expected),
+            bucket_lifecycle_lock_fence: Some(NamespaceLockFence::lost_for_test()),
+            ..Default::default()
+        };
+        assert!(matches!(
+            ensure_multipart_bucket_lifecycle_lock_held("bucket", "object", &lost),
+            Err(StorageError::NamespaceLockQuorumUnavailable { .. })
+        ));
+    }
+
     struct SetupTypeGuard {
         previous: SetupType,
     }
@@ -3721,7 +3989,7 @@ mod tests {
         // A single page must never return more than max_uploads entries.
         let max_uploads = 2usize;
         let page = set_disks
-            .list_multipart_uploads(bucket, object, None, None, None, max_uploads)
+            .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, max_uploads, None)
             .await
             .expect("list should succeed");
         assert_eq!(
@@ -3738,7 +4006,7 @@ mod tests {
 
         // Exact boundary: max_uploads == total must not falsely report truncation.
         let exact = set_disks
-            .list_multipart_uploads(bucket, object, None, None, None, total)
+            .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, total, None)
             .await
             .expect("list should succeed");
         assert_eq!(exact.uploads.len(), total, "exact boundary must return every upload");
@@ -3756,7 +4024,15 @@ mod tests {
         let mut pages = 0usize;
         loop {
             let page = set_disks
-                .list_multipart_uploads(bucket, object, key_marker.clone(), upload_id_marker.clone(), None, 1)
+                .list_multipart_uploads_for_incarnation(
+                    bucket,
+                    object,
+                    key_marker.clone(),
+                    upload_id_marker.clone(),
+                    None,
+                    1,
+                    None,
+                )
                 .await
                 .expect("list should succeed");
             assert!(page.uploads.len() <= 1, "max_uploads=1 must never return more than one upload");
@@ -3802,7 +4078,7 @@ mod tests {
         expected.sort();
 
         let all = set_disks
-            .list_multipart_uploads(bucket, "", None, None, None, 1000)
+            .list_multipart_uploads_for_incarnation(bucket, "", None, None, None, 1000, None)
             .await
             .expect("bucket-wide multipart listing should succeed");
         let listed = all
@@ -3814,14 +4090,14 @@ mod tests {
         assert!(!all.is_truncated);
 
         let logs = set_disks
-            .list_multipart_uploads(bucket, "logs/", None, None, None, 1000)
+            .list_multipart_uploads_for_incarnation(bucket, "logs/", None, None, None, 1000, None)
             .await
             .expect("prefix multipart listing should succeed");
         assert_eq!(logs.uploads.len(), 3);
         assert!(logs.uploads.iter().all(|upload| upload.object.starts_with("logs/")));
 
         let exact = set_disks
-            .list_multipart_uploads(bucket, "logs/a.bin", None, None, None, 1000)
+            .list_multipart_uploads_for_incarnation(bucket, "logs/a.bin", None, None, None, 1000, None)
             .await
             .expect("exact-key multipart listing should remain supported");
         assert_eq!(exact.uploads.len(), 2);
@@ -3851,7 +4127,15 @@ mod tests {
         let mut listed = Vec::new();
         for _ in 0..expected.len() {
             let page = set_disks
-                .list_multipart_uploads(bucket, "logs/", key_marker.clone(), upload_id_marker.clone(), None, 1)
+                .list_multipart_uploads_for_incarnation(
+                    bucket,
+                    "logs/",
+                    key_marker.clone(),
+                    upload_id_marker.clone(),
+                    None,
+                    1,
+                    None,
+                )
                 .await
                 .expect("multipart page should succeed");
             assert_eq!(page.uploads.len(), 1);
@@ -3868,7 +4152,7 @@ mod tests {
         assert_eq!(listed, expected);
 
         let key_only = set_disks
-            .list_multipart_uploads(bucket, "logs/", Some("logs/a.bin".to_string()), None, None, 1000)
+            .list_multipart_uploads_for_incarnation(bucket, "logs/", Some("logs/a.bin".to_string()), None, None, 1000, None)
             .await
             .expect("key-only marker should succeed");
         assert_eq!(
@@ -3881,7 +4165,7 @@ mod tests {
         );
 
         let upload_only = set_disks
-            .list_multipart_uploads(bucket, "logs/", None, Some(expected[0].1.clone()), None, 1000)
+            .list_multipart_uploads_for_incarnation(bucket, "logs/", None, Some(expected[0].1.clone()), None, 1000, None)
             .await
             .expect("an upload marker without a key marker should be ignored");
         assert_eq!(upload_only.uploads.len(), expected.len());
@@ -3909,7 +4193,7 @@ mod tests {
         }
 
         let first = set_disks
-            .list_multipart_uploads(bucket, "logs/", None, None, Some("/".to_string()), 2)
+            .list_multipart_uploads_for_incarnation(bucket, "logs/", None, None, Some("/".to_string()), 2, None)
             .await
             .expect("delimiter multipart listing should succeed");
         assert_eq!(first.uploads.len(), 1);
@@ -3920,13 +4204,14 @@ mod tests {
         assert!(first.next_upload_id_marker.is_none());
 
         let second = set_disks
-            .list_multipart_uploads(
+            .list_multipart_uploads_for_incarnation(
                 bucket,
                 "logs/",
                 first.next_key_marker,
                 first.next_upload_id_marker,
                 Some("/".to_string()),
                 2,
+                None,
             )
             .await
             .expect("delimiter continuation should succeed");
@@ -3936,7 +4221,7 @@ mod tests {
         assert!(!second.is_truncated);
 
         let exact_boundary = set_disks
-            .list_multipart_uploads(bucket, "logs/", None, None, Some("/".to_string()), 4)
+            .list_multipart_uploads_for_incarnation(bucket, "logs/", None, None, Some("/".to_string()), 4, None)
             .await
             .expect("delimiter exact boundary should succeed");
         assert_eq!(exact_boundary.uploads.len(), 2);
@@ -3944,6 +4229,48 @@ mod tests {
         assert!(!exact_boundary.is_truncated);
     }
 
+    #[tokio::test]
+    async fn list_multipart_uploads_hides_uploads_from_another_incarnation() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-incarnation-list-bucket";
+        let object = "object";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let incarnation = Uuid::new_v4();
+        let (fence, _loss_handle) = NamespaceLockFence::loss_handle_for_test();
+        let current = set_disks
+            .new_multipart_upload(
+                bucket,
+                object,
+                &ObjectOptions {
+                    expected_bucket_incarnation_id: Some(incarnation),
+                    bucket_lifecycle_lock_fence: Some(fence),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("current multipart upload should be created");
+        set_disks
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("legacy multipart upload should be created");
+
+        let unscoped = set_disks
+            .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, 1000, None)
+            .await
+            .expect("unscoped multipart listing should succeed");
+        assert_eq!(unscoped.uploads.len(), 2);
+
+        let scoped = set_disks
+            .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, 1000, Some(incarnation))
+            .await
+            .expect("incarnation-scoped multipart listing should succeed");
+        assert_eq!(scoped.uploads.len(), 1);
+        assert_eq!(scoped.uploads[0].upload_id, current.upload_id);
+    }
+
     /// Recursively collect every file named `file_name` under the multipart
     /// staging bucket on a single disk. Used to observe whether a failed commit
     /// left the per-part metadata intact for a retry.
@@ -4425,7 +4752,7 @@ mod tests {
 
         async fn upload_is_listed(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, upload_id: &str) -> bool {
             let page = set_disks
-                .list_multipart_uploads(bucket, object, None, None, None, 1000)
+                .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, 1000, None)
                 .await
                 .expect("listing multipart uploads should succeed");
             page.uploads.iter().any(|u| u.upload_id == upload_id)

crates/ecstore/src/set_disk/replication.rs

@@ -37,7 +37,10 @@ impl RestoreCleanupIdentity {
     }
 
     fn matches_file_info(&self, fi: &FileInfo, expected_etag: &str) -> bool {
-        self.version_id == fi.version_id
+        // Normalize the nil version on both sides: a versioning-suspended object
+        // is `Some(Uuid::nil())` on one and `None` on the other, so a raw compare
+        // reports every suspended restore as "changed before finalization".
+        self.version_id.filter(|version_id| !version_id.is_nil()) == fi.version_id.filter(|version_id| !version_id.is_nil())
             && self.data_dir == fi.data_dir
             && self.mod_time == fi.mod_time
             && self.size == fi.size