fix(asg): guard release_instances against double capacity decrement

After commit 641a033 introduced the IN_PROGRESS-then-COMPLETED return
flow, ASG partial-return tests started seeing DesiredCapacity drop by 2
(e.g. 4→2) when only 1 instance was returned (expected 4→3).

Root cause: release_instances calls detach_instances with
ShouldDecrementDesiredCapacity=True, which atomically decrements
DesiredCapacity. If the method is invoked a second time for the same
instance before the instance is confirmed terminated (possible under the
new IN_PROGRESS polling window), the decrement fires again on an already-
detached instance.

Fix: add _filter_asg_members() to query describe_auto_scaling_instances
before detaching. Only instances that are currently attached to the ASG
are included in the detach call. Instances already detached are skipped
(no capacity decrement), but are still forwarded to
terminate_instances_with_fallback so they are cleaned up.

Also corrects the fallback live_desired calculation to use
len(instances_to_detach) instead of len(instance_ids), keeping the
capacity arithmetic consistent when a subset is detached.

Adds a moto regression test (test_asg_partial_return_idempotent_capacity_
not_double_decremented) that calls release_hosts twice for the same
instance and asserts DesiredCapacity stays at N-1, not N-2.

Author: fgogolli

src/orb/providers/aws/infrastructure/handlers/asg/capacity_manager.py

@@ -172,8 +172,35 @@ def release_instances(
             self._cleanup_on_zero_capacity("asg", asg_name)
             return
 
+        # Guard against double-execution: only detach instances that are currently
+        # members of this ASG.  If release_instances is called a second time for the
+        # same set (e.g. due to an IN_PROGRESS status retry or a race), the instances
+        # are already detached and this check prevents a second DesiredCapacity decrement.
+        instances_to_detach = self._filter_asg_members(asg_name, instance_ids)
+        if not instances_to_detach:
+            self._logger.info(
+                "ASG %s: all %d instance(s) already detached — skipping detach and capacity decrement",
+                asg_name,
+                len(instance_ids),
+            )
+            # Instances may still be running (standalone after a prior partial detach);
+            # terminate them directly so the return request can complete.
+            self._aws_ops.terminate_instances_with_fallback(
+                instance_ids, self._request_adapter, f"ASG {asg_name} instances (already detached)"
+            )
+            return
+
+        skipped = [i for i in instance_ids if i not in instances_to_detach]
+        if skipped:
+            self._logger.info(
+                "ASG %s: %d instance(s) already detached (skipping): %s",
+                asg_name,
+                len(skipped),
+                skipped,
+            )
+
         # Detach instances (API limit: 50 per call; use 20 for safety)
-        for chunk in self._chunk_list(instance_ids, 20):
+        for chunk in self._chunk_list(instances_to_detach, 20):
             self._retry_with_backoff(
                 self._aws_client.autoscaling_client.detach_instances,
                 operation_type="critical",
@@ -182,7 +209,7 @@ def release_instances(
                 ShouldDecrementDesiredCapacity=True,
             )
             self._logger.debug("Detached chunk from ASG %s: %s", asg_name, chunk)
-        self._logger.info("Detached instances from ASG %s: %s", asg_name, instance_ids)
+        self._logger.info("Detached instances from ASG %s: %s", asg_name, instances_to_detach)
 
         # Re-describe the ASG to get the live DesiredCapacity after detach, since
         # ShouldDecrementDesiredCapacity=True may have already decremented it in AWS
@@ -201,7 +228,7 @@ def release_instances(
                 asg_name,
                 exc,
             )
-            live_desired = max(0, asg_details["DesiredCapacity"] - len(instance_ids))
+            live_desired = max(0, asg_details["DesiredCapacity"] - len(instances_to_detach))
 
         new_capacity = max(0, live_desired)
 
@@ -228,6 +255,40 @@ def release_instances(
     # Private helpers
     # ------------------------------------------------------------------
 
+    def _filter_asg_members(self, asg_name: str, instance_ids: list[str]) -> list[str]:
+        """Return the subset of instance_ids that are currently attached to asg_name.
+
+        Used as an idempotency guard in release_instances: if an instance has already
+        been detached (e.g. by a prior call), it is excluded so that
+        ShouldDecrementDesiredCapacity=True is not applied a second time.
+        """
+        if not instance_ids:
+            return []
+        try:
+            response = self._retry_with_backoff(
+                self._aws_client.autoscaling_client.describe_auto_scaling_instances,
+                operation_type="read_only",
+                InstanceIds=instance_ids,
+            )
+            attached_ids = {
+                entry["InstanceId"]
+                for entry in response.get("AutoScalingInstances", [])
+                if entry.get("AutoScalingGroupName") == asg_name
+            }
+            return [iid for iid in instance_ids if iid in attached_ids]
+        except Exception as exc:
+            self._logger.warning(
+                "Failed to verify ASG %s membership for instances %s; "
+                "proceeding with all instances to avoid leaving them running: %s",
+                asg_name,
+                instance_ids,
+                exc,
+            )
+            # On error, be conservative: attempt detach for all instances.
+            # A "not a member" error from detach_instances is better than
+            # skipping a needed capacity decrement.
+            return list(instance_ids)
+
     def _call_delete_asg(self, asg_name: str) -> None:
         """Invoke the registered delete-ASG callback, or fall back to direct deletion."""
         if self._delete_asg_fn is not None:

tests/providers/aws/moto/test_partial_return.py

@@ -467,6 +467,77 @@ def test_asg_partial_return_decrements_desired_capacity(self, moto_vpc_resources
         assert resp["AutoScalingGroups"], "ASG should still exist after partial release"
         assert resp["AutoScalingGroups"][0]["DesiredCapacity"] == 1
 
+    def test_asg_partial_return_idempotent_capacity_not_double_decremented(self, moto_vpc_resources):
+        """Calling release_hosts twice for the same instance must NOT decrement capacity twice.
+
+        Regression guard for the IN_PROGRESS status flow introduced in commit 641a03332:
+        with the return request staying IN_PROGRESS after termination is accepted, a second
+        release_hosts call (any re-entrant path) must leave DesiredCapacity at N-1, not N-2.
+        """
+        subnet_id = moto_vpc_resources["subnet_ids"][0]
+        sg_id = moto_vpc_resources["sg_id"]
+
+        aws_client = _make_capacity_aws_client()
+        logger = _make_logger()
+        config_port = _make_capacity_config_port()
+        lt_manager = _make_capacity_lt_manager(aws_client)
+        aws_ops = AWSOperations(aws_client, logger, config_port)
+        handler = ASGHandler(
+            aws_client=aws_client,
+            logger=logger,
+            aws_ops=aws_ops,
+            launch_template_manager=lt_manager,
+            config_port=config_port,
+        )
+
+        template = AWSTemplate(
+            template_id="tpl-asg-idempotent",
+            name="test-asg-idempotent",
+            provider_api="ASG",
+            machine_types={"t3.micro": 1},
+            image_id="ami-12345678",
+            max_instances=4,
+            price_type="ondemand",
+            subnet_ids=[subnet_id],
+            security_group_ids=[sg_id],
+        )
+        request = _make_capacity_request("req-asg-idempotent-001", requested_count=4)
+        result = handler.acquire_hosts(request, template)
+        assert result["success"] is True
+        asg_name = result["resource_ids"][0]
+
+        # Provision 4 instances and attach them to the ASG
+        ec2 = aws_client.ec2_client
+        asg = aws_client.autoscaling_client
+        run_resp = ec2.run_instances(
+            ImageId="ami-12345678",
+            MinCount=4,
+            MaxCount=4,
+            InstanceType="t3.micro",
+            SubnetId=subnet_id,
+        )
+        instance_ids = [i["InstanceId"] for i in run_resp["Instances"]]
+        asg.attach_instances(InstanceIds=instance_ids, AutoScalingGroupName=asg_name)
+        asg.update_auto_scaling_group(AutoScalingGroupName=asg_name, DesiredCapacity=4)
+
+        resource_mapping = {instance_ids[0]: (asg_name, 4)}
+
+        # First call — normal deprovisioning: capacity 4 → 3
+        handler.release_hosts([instance_ids[0]], resource_mapping=resource_mapping)
+        resp = asg.describe_auto_scaling_groups(AutoScalingGroupNames=[asg_name])
+        assert resp["AutoScalingGroups"][0]["DesiredCapacity"] == 3, (
+            "First release_hosts call should decrement capacity from 4 to 3"
+        )
+
+        # Second call — simulates re-entrancy (IN_PROGRESS retry scenario):
+        # instance is already detached; capacity must NOT drop to 2.
+        handler.release_hosts([instance_ids[0]], resource_mapping=resource_mapping)
+        resp = asg.describe_auto_scaling_groups(AutoScalingGroupNames=[asg_name])
+        assert resp["AutoScalingGroups"][0]["DesiredCapacity"] == 3, (
+            "Second release_hosts call for already-detached instance must NOT decrement "
+            "capacity again (expected 3, idempotency guard should prevent 3 → 2)"
+        )
+
     def test_ec2_fleet_maintain_partial_return_decrements_target_capacity(self, moto_vpc_resources):
         """release_hosts with resource_mapping for 1 unit calls modify_fleet with TotalTargetCapacity=1.