fix(scanner): fence backlog writes against stale pool membership (#8799)

* perf(s3): borrow options during delete preflight

Pass the existing request options by reference to the metadata preflight instead of cloning them solely for that borrow. Keep the getter's mutable snapshot copy and the later owned-delete copy.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(scanner): fence backlog writes against stale pool membership

Reject native scanner publications unless their captured writable pool cohort
matches the fenced durable topology. A smaller startup view must not replace
an expanded membership commit with its older stable rollback.

Keep single-replica test injection fenced by the complete current cohort and
cover rejected startup claims, unchanged CAS revisions, and recovery of the
acknowledged expanded commit.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(ecstore): use current cohort for healthy scanner CAS writes

Keep the writer's cached topology stale while explicitly capturing the
surviving membership for healthy-target publication and CAS checks. Also
assert that the old cohort rejects healthy-target writes without publishing.

Preserve retiring-source rejection, payload and ETag checks, and canceled
source membership repair without changing production admission.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/core/pools.rs

@@ -12781,6 +12781,19 @@ impl ECStore {
         let (pool_meta_guard, snapshot) = self
             .acquire_pool_meta_read_guard(&save_guard, "target capacity admission failed")
             .await?;
+        // A scanner publication must cover the complete durable membership.
+        // A smaller startup view could otherwise overwrite an expanded commit
+        // proof with its older stable rollback, even though every target is writable.
+        if matches!(admission, DecommissionCapacityAdmission::ScannerBacklog)
+            && !snapshot
+                .pools
+                .iter()
+                .enumerate()
+                .filter_map(|(index, _)| snapshot.scanner_pause_backlog_pool_writable(index).then_some(index))
+                .eq(target_pool_indices.iter().copied())
+        {
+            return Err(Error::SlowDown);
+        }
         for target_pool_index in target_pool_indices.iter().copied() {
             ensure_external_decommission_target_admission(&snapshot, target_pool_index, admission)?;
         }

crates/ecstore/src/core/pools_test.rs

@@ -7891,10 +7891,27 @@ mod decommission_lock_order_tests {
                 .expect("read the entire retained source");
             assert_eq!(actual, body);
 
+            // A new publication captures the surviving membership even while
+            // the storage handle retains its old node-local metadata.
+            let surviving_pools: Arc<[usize]> = Arc::from([1, 2]);
             for set in &writer.pools[2].disk_set {
+                let stale = writer
+                    .save_scanner_pause_backlog_replica(
+                        2,
+                        set.set_index,
+                        b"late stale-cohort target update".to_vec(),
+                        crate::storage_api_contracts::object::HTTPPreconditions {
+                            if_none_match: Some("*".to_string()),
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect_err("the stale cohort must also fence publication on a healthy target");
+                assert!(matches!(stale, crate::error::Error::SlowDown));
+
                 let target_body = format!("surviving native scanner set {}", set.set_index).into_bytes();
                 let committed = writer
-                    .save_scanner_pause_backlog_replica(
+                    .save_scanner_pause_backlog_replica_for_membership(
                         2,
                         set.set_index,
                         target_body.clone(),
@@ -7902,11 +7919,12 @@ mod decommission_lock_order_tests {
                             if_none_match: Some("*".to_string()),
                             ..Default::default()
                         },
+                        Arc::clone(&surviving_pools),
                     )
                     .await
-                    .expect("every reserved healthy target set must still accept scanner replicas");
+                    .expect("the current surviving cohort must admit every reserved healthy target set");
                 let conflict = writer
-                    .save_scanner_pause_backlog_replica(
+                    .save_scanner_pause_backlog_replica_for_membership(
                         2,
                         set.set_index,
                         b"must not bypass CAS".to_vec(),
@@ -7914,6 +7932,7 @@ mod decommission_lock_order_tests {
                             if_match: Some("stale-native-revision".to_string()),
                             ..Default::default()
                         },
+                        Arc::clone(&surviving_pools),
                     )
                     .await
                     .expect_err("capacity admission must retain the native writer's CAS");
@@ -7930,6 +7949,10 @@ mod decommission_lock_order_tests {
                     .expect("read the complete native target");
                 assert_eq!(actual, target_body);
             }
+            assert!(
+                writer.pool_meta.read().await.pools[0].decommission.is_none(),
+                "valid captured membership must work without refreshing the writer's stale snapshot"
+            );
             for (pool_index, set_index) in [(writer.pools.len(), 0), (0, writer.pools[0].disk_set.len())] {
                 assert!(matches!(
                     writer

crates/ecstore/src/store/object.rs

@@ -3524,14 +3524,16 @@ impl ECStore {
         data: Vec<u8>,
         preconditions: crate::storage_api_contracts::object::HTTPPreconditions,
     ) -> Result<ObjectInfo> {
-        self.save_scanner_pause_backlog_replica_for_membership(
-            pool_index,
-            set_index,
-            data,
-            preconditions,
-            Arc::from([pool_index]),
-        )
-        .await
+        let writable_pool_indices = self
+            .scanner_pause_backlog_writable_set_disks()
+            .await
+            .into_iter()
+            .map(|set| set.pool_index)
+            .collect::<std::collections::BTreeSet<_>>()
+            .into_iter()
+            .collect();
+        self.save_scanner_pause_backlog_replica_for_membership(pool_index, set_index, data, preconditions, writable_pool_indices)
+            .await
     }
 
     /// Publish a native scanner replica without allowing stale pool selection

crates/scanner/src/scanner/backlog.rs

@@ -3184,60 +3184,42 @@ mod tests {
 
     #[test]
     #[serial_test::serial]
-    fn native_writer_expansion_rechecks_membership_after_seeding() {
+    fn native_writer_expansion_rejects_an_incomplete_startup_membership() {
         run_native_retirement_test(async || {
-            use crate::storage_api::owner::NativeScannerPauseBacklogWriteFault;
-
             let (_root, store, original) = native_expanded_backlog_store(1, false).await;
-            // Model a node-local writable view learning about another pool while
-            // the already selected seed fan-out is in flight. Durable admission
-            // still uses the complete pool metadata throughout the test.
             let later_member = store.pool_meta.write().await.pools.pop().expect("third pool visibility");
             let before = load_scanner_pause_backlog(Arc::clone(&store))
                 .await
                 .expect("initial writable view");
             assert_eq!(before.replica_count, 4);
             assert_eq!(before.healthy_replicas, 2);
-            let barrier =
-                NativeScannerPauseBacklogWriteFault::pause_after_write(Arc::clone(&store.pools[1].disk_set[1]), "publish");
-            let late_failure =
-                NativeScannerPauseBacklogWriteFault::fail_before_write(Arc::clone(&store.pools[2].disk_set[0]), "publish", 1);
             let now = unix_now().saturating_add(60);
-            let writer_store = Arc::clone(&store);
-            let writer = tokio::spawn(async move { ScannerPauseBacklogController::claim(writer_store, now).await });
-            tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
-                .await
-                .expect("seed publication must reach the native barrier");
+            let result = ScannerPauseBacklogController::claim(Arc::clone(&store), now).await;
             store.pool_meta.write().await.pools.push(later_member);
-            barrier.release();
-            let error = tokio::time::timeout(Duration::from_secs(30), writer)
-                .await
-                .expect("writer must finish after the seed barrier")
-                .expect("writer task must not panic")
+            let error = result
                 .err()
-                .expect("a newly visible unseeded member must block the next commit");
-            assert!(error.contains("failed to stabilize its last committed generation"), "{error}");
-            drop(barrier);
-            drop(late_failure);
+                .expect("durable membership must reject the incomplete startup view");
+            assert!(error.contains("commit did not reach every surviving set"), "{error}");
 
             let loaded = load_scanner_pause_backlog(Arc::clone(&store))
                 .await
-                .expect("the old native authority must survive membership growth during seeding");
+                .expect("the old authority must survive a rejected startup publication");
             assert_eq!(loaded.ledger, original);
             assert_eq!(loaded.authoritative_commit, before.authoritative_commit);
             assert_eq!(loaded.replica_count, 6);
-            assert_eq!(loaded.healthy_replicas, 4);
-            for set in &store.pools[2].disk_set {
-                assert!(matches!(
-                    read_scanner_pause_backlog_replica(Arc::clone(set)).await.state,
-                    ScannerPauseBacklogReplicaState::Missing
-                ));
+            assert_eq!(loaded.healthy_replicas, 2);
+            for pool in store.pools.iter().skip(1) {
+                for set in &pool.disk_set {
+                    assert!(matches!(
+                        read_scanner_pause_backlog_replica(Arc::clone(set)).await.state,
+                        ScannerPauseBacklogReplicaState::Missing
+                    ));
+                }
             }
-
             let expected = claim_scanner_pause_backlog_writer(&original, now).expect("retry writer generation");
             ScannerPauseBacklogController::claim(Arc::clone(&store), now)
                 .await
-                .expect("retry must seed the newly visible members before committing");
+                .expect("the complete membership seeds every new member before committing");
             assert_current_native_writer_ledger(&store, &expected).await;
             shutdown_native_retirement_store(store).await;
         });
@@ -3302,6 +3284,68 @@ mod tests {
         });
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn native_writer_stale_cohort_cannot_roll_back_expansion_commit() {
+        run_native_retirement_test(async || {
+            let (_root, store) = native_pause_backlog_store(3, 1).await;
+            let now = unix_now();
+            let old = ScannerPauseBacklogController::claim(Arc::clone(&store), now)
+                .await
+                .expect("initial stable cohort")
+                .loaded;
+            let next = claim_scanner_pause_backlog_writer(&old.ledger, now + 1).expect("next native writer");
+            let members = scanner_pause_backlog_replica_ids(&old.replicas);
+            write_scanner_pause_backlog_record(
+                Arc::clone(&store),
+                &old,
+                commit_scanner_pause_backlog_record(Some(&old.ledger), &next, &members),
+            )
+            .await
+            .expect("complete expanded membership commit before stabilization");
+            let pending = load_scanner_pause_backlog(Arc::clone(&store))
+                .await
+                .expect("expanded commit must be authoritative");
+            assert_eq!(pending.ledger, next);
+            let mut before = Vec::new();
+            for pool in &store.pools {
+                before.push(native_replica_bytes(&pool.disk_set[0]).await);
+            }
+
+            // Startup can expose an older node-local membership while durable
+            // pool metadata already includes every configured pool. That view
+            // cannot recognize the expanded commit and selects its rollback.
+            let later_member = store.pool_meta.write().await.pools.pop().expect("last pool visibility");
+            let stale = load_scanner_pause_backlog(Arc::clone(&store))
+                .await
+                .expect("old view selects its stable rollback");
+            assert_eq!(stale.ledger, old.ledger);
+            assert!(stale.authoritative_commit.is_none());
+            let stale_members = scanner_pause_backlog_replica_ids(&stale.replicas);
+            let result = write_scanner_pause_backlog_record(
+                Arc::clone(&store),
+                &stale,
+                stable_scanner_pause_backlog_record(&stale.ledger, None, &stale_members),
+            )
+            .await;
+            store.pool_meta.write().await.pools.push(later_member);
+            result.expect_err("a smaller stale cohort must not overwrite the expanded commit proof");
+            for (pool, expected) in store.pools.iter().zip(before) {
+                assert_eq!(
+                    native_replica_bytes(&pool.disk_set[0]).await,
+                    expected,
+                    "rejected publication preserves each payload and CAS revision"
+                );
+            }
+            let stabilized = stabilize_scanner_pause_backlog(Arc::clone(&store), &pending)
+                .await
+                .expect("the acknowledged expanded commit remains recoverable");
+            assert!(stabilized.all_current_replicas_stable());
+            assert_current_native_writer_ledger(&store, &next).await;
+            shutdown_native_retirement_store(store).await;
+        });
+    }
+
     #[test]
     #[serial_test::serial]
     fn native_writer_stale_cohort_cannot_stabilize_survivors_after_retirement() {

rustfs/src/app/object/delete.rs

@@ -1137,8 +1137,7 @@ impl DefaultObjectUsecase {
         let replicate_force_delete = force_delete && !replica && has_active_delete_rule(&delete_config_snapshot, &key);
         let mut force_delete_intent = None;
 
-        let get_opts = opts.clone();
-        let existing_object_info = match store.get_object_info_for_delete(&bucket, &key, &get_opts).await {
+        let existing_object_info = match store.get_object_info_for_delete(&bucket, &key, &opts).await {
             Ok(obj_info) => Some(obj_info),
             Err(err) => {
                 // If object not found, allow deletion to proceed (will return 204 No Content)