fix: Let every device settle before a bulk failure surfaces

Applying config across devices used a plain gather, so one device raising
propagated immediately while its siblings carried on issuing SetConfig to real
hardware with nobody awaiting them. That is the same hazard the capture path
already guarded against, and it matters more here because these calls write.

Both paths now share one helper that runs every device to completion before
surfacing the first failure.

Author: jfmlima

packages/core/src/core/use_cases/bulk_operations.py

@@ -1,13 +1,16 @@
 import asyncio
+from collections.abc import Coroutine, Iterable
 from datetime import UTC, datetime
-from typing import Any
+from typing import Any, TypeVar
 
 from ..domain.entities.config_snapshot import DeviceSnapshot
 from ..domain.entities.exceptions import BulkOperationError
 from ..domain.value_objects.action_result import ActionResult
 from ..gateways.device import DeviceGateway
 from .capture_device_config import CaptureDeviceConfig
 
+T = TypeVar("T")
+
 
 def _unique(device_ips: list[str]) -> list[str]:
     """The given IPs in order, without repeats.
@@ -20,6 +23,23 @@ def _unique(device_ips: list[str]) -> list[str]:
     return list(dict.fromkeys(device_ips))
 
 
+async def _gather_settled(coroutines: Iterable[Coroutine[Any, Any, T]]) -> list[T]:
+    """Run every device to completion, in order, then surface the first failure.
+
+    Plain ``asyncio.gather`` raises the moment one device fails and does not
+    cancel its siblings, leaving them issuing requests to real hardware with
+    nobody awaiting them. Letting every task settle first keeps one bad device
+    from doing that.
+    """
+    results = await asyncio.gather(*coroutines, return_exceptions=True)
+    settled: list[T] = []
+    for result in results:
+        if isinstance(result, BaseException):
+            raise result
+        settled.append(result)
+    return settled
+
+
 class BulkOperationsUseCase:
 
     def __init__(
@@ -116,7 +136,9 @@ async def export_bulk_config(
             Dictionary containing export metadata and device configurations
         """
         targets = _unique(device_ips)
-        snapshots = await self._capture_all(targets, component_types)
+        snapshots = await _gather_settled(
+            self._capture_device(ip, component_types) for ip in targets
+        )
 
         return {
             "export_metadata": {
@@ -131,26 +153,6 @@ async def export_bulk_config(
             },
         }
 
-    async def _capture_all(
-        self, device_ips: list[str], component_types: list[str]
-    ) -> list[DeviceSnapshot | None]:
-        """Capture every device concurrently, in the order they were asked for.
-
-        Gathering with ``return_exceptions`` lets every capture finish before a
-        failure surfaces, so one bad device does not leave the others running
-        against real hardware with nobody awaiting them.
-        """
-        results = await asyncio.gather(
-            *(self._capture_device(ip, component_types) for ip in device_ips),
-            return_exceptions=True,
-        )
-        snapshots: list[DeviceSnapshot | None] = []
-        for result in results:
-            if isinstance(result, BaseException):
-                raise result
-            snapshots.append(result)
-        return snapshots
-
     async def _capture_device(
         self, device_ip: str, component_types: list[str]
     ) -> DeviceSnapshot | None:
@@ -183,11 +185,9 @@ async def apply_bulk_config(
         Returns:
             List of action results
         """
-        per_device = await asyncio.gather(
-            *(
-                self._apply_device_config(device_ip, component_type, config)
-                for device_ip in _unique(device_ips)
-            )
+        per_device = await _gather_settled(
+            self._apply_device_config(device_ip, component_type, config)
+            for device_ip in _unique(device_ips)
         )
         return [result for results in per_device for result in results]
 

packages/core/tests/unit/use_cases/test_bulk_operations.py

@@ -1013,3 +1013,28 @@ async def test_it_exports_a_repeated_device_only_once(
 
         assert mock_device_gateway.get_device_status.await_count == 1
         assert result["export_metadata"]["total_devices"] == 1
+
+    async def test_it_lets_every_apply_finish_before_a_failure_surfaces(
+        self, use_case, mock_device_gateway
+    ):
+        # Same rule as the capture path: a device that raises must not leave its
+        # siblings still writing to hardware with nobody awaiting them.
+        finished = []
+
+        async def get_component_keys(device_ip, component_type):
+            if device_ip == "192.168.1.100":
+                raise RuntimeError("boom")
+            await asyncio.sleep(0.05)
+            finished.append(device_ip)
+            return []
+
+        mock_device_gateway.get_component_keys = AsyncMock(
+            side_effect=get_component_keys
+        )
+
+        with pytest.raises(RuntimeError, match="boom"):
+            await use_case.apply_bulk_config(
+                ["192.168.1.100", "192.168.1.101"], "switch", {}
+            )
+
+        assert finished == ["192.168.1.101"]