fix(ecstore): keep decommission multipart on reserved target (#7937)

Route capacity-owned unversioned multipart uploads through the durable decommission target reservation and fail closed when the reservation lease has expired.

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() {

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() {

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));
         }