diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 294f53ebb..e3825d85a 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/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)?; } diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 5418522cf..ab950fb68 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/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 diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 11c349ac5..9c0549bf4 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -3524,14 +3524,16 @@ impl ECStore { data: Vec, preconditions: crate::storage_api_contracts::object::HTTPPreconditions, ) -> Result { - 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::>() + .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 diff --git a/crates/scanner/src/scanner/backlog.rs b/crates/scanner/src/scanner/backlog.rs index f5934de41..e7ce66801 100644 --- a/crates/scanner/src/scanner/backlog.rs +++ b/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() { diff --git a/rustfs/src/app/object/delete.rs b/rustfs/src/app/object/delete.rs index d828593e5..c9e84e121 100644 --- a/rustfs/src/app/object/delete.rs +++ b/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)