diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 071f98277..f325c10c7 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -12446,10 +12446,14 @@ impl ECStore { .get(idx) .and_then(|pool| pool.decommission.as_ref()) .and_then(|info| info.capacity_reservation.as_ref()) - .filter(|reservation| reservation.lease_active_at(OffsetDateTime::now_utc())) else { return Ok(None); }; + if !reservation.lease_active_at(OffsetDateTime::now_utc()) { + return Err(decommission_capacity_blocked_error( + "decommission capacity reservation lease is not active", + )); + } Ok(Some(DecommissionCapacityOwner { source_pool_index: idx, operation_id: reservation.operation_id, @@ -20057,6 +20061,45 @@ mod tests { ); } + #[tokio::test] + #[serial_test::serial] + async fn expired_decommission_capacity_lease_requires_recovery_before_entry() { + let (_temp_dirs, store, _other_store) = + crate::services::rebalance::test_three_pool_stores_with_isolated_node_contexts(None).await; + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + set_decommission_capacity_info_overrides_for_test( + store.id, + vec![vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 1, 1), + DecommissionPoolCapacityInfo::for_test(1, layout, 2, 2, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 2, 2, 0), + ]], + ); + store + .save_current_pool_meta_for_decommission_start(&[0], Vec::new()) + .await + .expect("activate the expired lease reservation"); + let generation = store + .active_decommission_generation(0) + .await + .expect("read the active decommission generation"); + { + let mut pool_meta = store.pool_meta.write().await; + let reservation = pool_meta.pools[0] + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .expect("the expired lease reservation should exist"); + reservation.expires_at = OffsetDateTime::now_utc() - Duration::seconds(1); + } + + let err = store + .decommission_capacity_owner_for_worker(0, generation) + .await + .expect_err("an expired capacity lease must not fall back to ownerless migration"); + assert!(is_decommission_capacity_blocked_error(&err)); + } + #[tokio::test] #[serial_test::serial] async fn expired_capacity_owner_nonce_rejects_stale_put_and_multipart_without_consuming() { diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 76de24d8c..2c08f25a7 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/crates/ecstore/src/core/pools_test.rs @@ -2883,6 +2883,89 @@ mod decommission_lock_order_tests { assert!(upload_info.parts.is_empty()); } + #[test] + #[serial_test::serial] + fn unversioned_decommission_multipart_uses_reserved_target() { + run_large_stack_current_thread_async_test( + "unversioned-decommission-multipart-target", + unversioned_decommission_multipart_uses_reserved_target_case, + ); + } + + async fn unversioned_decommission_multipart_uses_reserved_target_case() { + let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await; + let bucket = test_bucket("unversioned-decommission-multipart-target"); + let object = "unversioned-existing-unreserved-target.bin"; + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let capacity_snapshot = || { + vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 1, 1), + DecommissionPoolCapacityInfo::for_test(1, layout, 2, 2, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 100, 100, 0), + ] + }; + + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create the unversioned multipart target bucket"); + let incarnation = store + .bucket_incarnation_id(&bucket) + .await + .expect("load the unversioned multipart target incarnation"); + + // Force the ordinary pool selector to choose pool 2. A capacity-owned + // upload must ignore this unreserved existing object and use pool 1. + let mut existing = PutObjReader::from_vec(b"existing unreserved target".to_vec()); + store.pools[2] + .put_object( + &bucket, + object, + &mut existing, + &ObjectOptions { + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }, + ) + .await + .expect("seed the unreserved target object"); + + set_decommission_capacity_info_overrides_for_test(store.id, vec![capacity_snapshot()]); + store + .save_current_pool_meta_for_decommission_start(&[0], Vec::new()) + .await + .expect("activate the unversioned multipart reservation"); + { + let pool_meta = store.pool_meta.read().await; + let targets = &pool_meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .expect("the unversioned multipart reservation should exist") + .targets; + assert_eq!(targets.len(), 1, "only pool 1 should be reserved"); + assert_eq!(targets[0].pool_index, 1); + } + + let owner = decommission_capacity_owner(&*store.pool_meta.read().await).with_mutation_id(uuid::Uuid::new_v4()); + let mut upload_opts = ObjectOptions { + data_movement: true, + src_pool_idx: 0, + versioned: false, + version_id: None, + mod_time: Some(time::OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(97)), + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }; + owner.apply_to(&mut upload_opts); + + let (_, target_pool_idx, _) = store + .handle_new_multipart_upload_with_pool_idx(&bucket, object, &upload_opts, None) + .await + .expect("unversioned decommission multipart must use the reserved target"); + assert_eq!(target_pool_idx, 1, "the reservation target must override the existing unreserved object"); + } + #[test] #[serial_test::serial] fn data_movement_multipart_abort_restart_reconciles_inflight_after_release_save_loss() { diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 2b14dc6a0..0cd7dd837 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -530,7 +530,11 @@ impl ECStore { return Ok((result, 0, opts.expected_bucket_incarnation_id)); } - if opts.data_movement && opts.version_id.is_some() { + let capacity_owner = DecommissionCapacityOwner::from_options(&opts); + if opts.data_movement && (opts.version_id.is_some() || capacity_owner.is_some()) { + // Capacity-owned decommission writes must remain on the target + // selected by the durable reservation, including unversioned + // objects whose ObjectOptions carry no version ID. let idx = self.select_data_movement_pool_idx(bucket, object, -1, &opts, false).await?; if idx == opts.src_pool_idx { return Err(StorageError::DataMovementOverwriteErr( @@ -542,12 +546,9 @@ impl ECStore { self.apply_decommission_target_mutation_fence(idx, object, &mut opts, mutation_fence) .await; let res = self - .run_decommission_capacity_temporary_mutation( - idx, - DecommissionCapacityOwner::from_options(&opts), - None, - || async { self.pools[idx].new_multipart_upload(bucket, object, &opts).await }, - ) + .run_decommission_capacity_temporary_mutation(idx, capacity_owner, None, || async { + self.pools[idx].new_multipart_upload(bucket, object, &opts).await + }) .await?; return Ok((res, idx, opts.expected_bucket_incarnation_id)); }