Fix concurrent AWS security group creation (#4399)

Handle InvalidGroup.Duplicate and
InvalidPermission.Duplicate errors when instances
in the same fleet provision at once.

Author: jvstme

src/dstack/_internal/core/backends/aws/resources.py

@@ -82,20 +82,34 @@ def create_security_group(
         create_security_group_kwargs = {}
         if vpc_id is not None:
             create_security_group_kwargs["VpcId"] = vpc_id
-        security_group = ec2_client.create_security_group(
-            Description="Generated by dstack",
-            GroupName=security_group_name,
-            TagSpecifications=[
-                {
-                    "ResourceType": "security-group",
-                    "Tags": [
-                        {"Key": "owner", "Value": "dstack"},
-                        {"Key": "dstack_project", "Value": project_id},
-                    ],
-                },
-            ],
-            **create_security_group_kwargs,
-        )
+        try:
+            security_group = ec2_client.create_security_group(
+                Description="Generated by dstack",
+                GroupName=security_group_name,
+                TagSpecifications=[
+                    {
+                        "ResourceType": "security-group",
+                        "Tags": [
+                            {"Key": "owner", "Value": "dstack"},
+                            {"Key": "dstack_project", "Value": project_id},
+                        ],
+                    },
+                ],
+                **create_security_group_kwargs,
+            )
+        except botocore.exceptions.ClientError as e:
+            if e.response.get("Error", {}).get("Code") != "InvalidGroup.Duplicate":
+                raise
+            # The group was created concurrently, e.g. by another instance provisioning
+            response = ec2_client.describe_security_groups(
+                Filters=describe_security_groups_filters
+            )
+            if not response.get("SecurityGroups"):
+                raise ComputeError(
+                    f"Security group {security_group_name} reported as duplicate"
+                    " but not found when describing it"
+                ) from e
+            security_group = response["SecurityGroups"][0]
     security_group_id = security_group["GroupId"]
 
     _add_ingress_security_group_rule_if_missing(
@@ -486,10 +500,16 @@ def _add_ingress_security_group_rule_if_missing(
 ) -> bool:
     if _rule_exists(rule, security_group.get("IpPermissions", [])):
         return False
-    ec2_client.authorize_security_group_ingress(
-        GroupId=security_group_id,
-        IpPermissions=[rule],
-    )
+    try:
+        ec2_client.authorize_security_group_ingress(
+            GroupId=security_group_id,
+            IpPermissions=[rule],
+        )
+    except botocore.exceptions.ClientError as e:
+        if e.response.get("Error", {}).get("Code") == "InvalidPermission.Duplicate":
+            # The rule was added concurrently
+            return False
+        raise
     return True
 
 
@@ -501,10 +521,16 @@ def _add_egress_security_group_rule_if_missing(
 ) -> bool:
     if _rule_exists(rule, security_group.get("IpPermissionsEgress", [])):
         return False
-    ec2_client.authorize_security_group_egress(
-        GroupId=security_group_id,
-        IpPermissions=[rule],
-    )
+    try:
+        ec2_client.authorize_security_group_egress(
+            GroupId=security_group_id,
+            IpPermissions=[rule],
+        )
+    except botocore.exceptions.ClientError as e:
+        if e.response.get("Error", {}).get("Code") == "InvalidPermission.Duplicate":
+            # The rule was added concurrently
+            return False
+        raise
     return True