CFn: DAG based deploy order (#10849)

localstack-core/localstack/config.py

@@ -1041,6 +1041,9 @@ def populate_edge_configuration(
 # Show exceptions for CloudFormation deploy errors
 CFN_VERBOSE_ERRORS = is_env_true("CFN_VERBOSE_ERRORS")
 
+# Allow fallback to previous template deployer implementation
+CFN_LEGACY_TEMPLATE_DEPLOYER = is_env_true("CFN_LEGACY_TEMPLATE_DEPLOYER")
+
 # Set the timeout to deploy each individual CloudFormation resource
 CFN_PER_RESOURCE_TIMEOUT = int(os.environ.get("CFN_PER_RESOURCE_TIMEOUT") or 300)
 
@@ -1104,6 +1107,7 @@ def use_custom_dns():
     "BOTO_WAITER_MAX_ATTEMPTS",
     "BUCKET_MARKER_LOCAL",
     "CFN_IGNORE_UNSUPPORTED_RESOURCE_TYPES",
+    "CFN_LEGACY_TEMPLATE_DEPLOYER",
     "CFN_VERBOSE_ERRORS",
     "CI",
     "CUSTOM_SSL_CERT_PATH",

localstack-core/localstack/services/cdk/resource_providers/cdk_metadata.py

@@ -70,7 +70,7 @@ def delete(
 
         return ProgressEvent(
             status=OperationStatus.SUCCESS,
-            resource_model=None,
+            resource_model=request.previous_state,
         )
 
     def update(

localstack-core/localstack/services/cloudformation/engine/changes.py

@@ -0,0 +1,18 @@
+from typing import Literal, Optional, TypedDict
+
+Action = str
+
+
+class ResourceChange(TypedDict):
+    Action: Action
+    LogicalResourceId: str
+    PhysicalResourceId: Optional[str]
+    ResourceType: str
+    Scope: list
+    Details: list
+    Replacement: Optional[Literal["False"]]
+
+
+class ChangeConfig(TypedDict):
+    Type: str
+    ResourceChange: ResourceChange

localstack-core/localstack/services/cloudformation/engine/resource_ordering.py

@@ -0,0 +1,109 @@
+from collections import OrderedDict
+
+from localstack.services.cloudformation.engine.changes import ChangeConfig
+from localstack.services.cloudformation.engine.parameters import StackParameter
+from localstack.services.cloudformation.engine.template_utils import get_deps_for_resource
+
+
+class NoResourceInStack(ValueError):
+    """Raised when we preprocess the template and do not find a resource"""
+
+    def __init__(self, logical_resource_id: str):
+        msg = f"Template format error: Unresolved resource dependencies [{logical_resource_id}] in the Resources block of the template"
+
+        super().__init__(msg)
+
+
+def order_resources(
+    resources: dict,
+    resolved_parameters: dict[str, StackParameter],
+    resolved_conditions: dict[str, bool],
+    reverse: bool = False,
+) -> OrderedDict:
+    """
+    Given a dictionary of resources, topologically sort the resources based on
+    inter-resource dependencies (e.g. usages of intrinsic functions).
+    """
+    nodes: dict[str, list[str]] = {}
+    for logical_resource_id, properties in resources.items():
+        nodes.setdefault(logical_resource_id, [])
+        deps = get_deps_for_resource(properties, resolved_conditions)
+        for dep in deps:
+            if dep in resolved_parameters:
+                # we only care about other resources
+                continue
+            nodes.setdefault(dep, [])
+            nodes[dep].append(logical_resource_id)
+
+    # implementation from https://dev.to/leopfeiffer/topological-sort-with-kahns-algorithm-3dl1
+    indegrees = {k: 0 for k in nodes.keys()}
+    for dependencies in nodes.values():
+        for dependency in dependencies:
+            indegrees[dependency] += 1
+
+    # Place all elements with indegree 0 in queue
+    queue = [k for k in nodes.keys() if indegrees[k] == 0]
+
+    sorted_logical_resource_ids = []
+
+    # Continue until all nodes have been dealt with
+    while len(queue) > 0:
+        # node of current iteration is the first one from the queue
+        curr = queue.pop(0)
+        sorted_logical_resource_ids.append(curr)
+
+        # remove the current node from other dependencies
+        for dependency in nodes[curr]:
+            indegrees[dependency] -= 1
+
+            if indegrees[dependency] == 0:
+                queue.append(dependency)
+
+    # check for circular dependencies
+    if len(sorted_logical_resource_ids) != len(nodes):
+        raise Exception("Circular dependency found.")
+
+    sorted_mapping = []
+    for logical_resource_id in sorted_logical_resource_ids:
+        if properties := resources.get(logical_resource_id):
+            sorted_mapping.append((logical_resource_id, properties))
+        else:
+            if (
+                logical_resource_id not in resolved_parameters
+                and logical_resource_id not in resolved_conditions
+            ):
+                raise NoResourceInStack(logical_resource_id)
+
+    if reverse:
+        sorted_mapping = sorted_mapping[::-1]
+    return OrderedDict(sorted_mapping)
+
+
+def order_changes(
+    given_changes: list[ChangeConfig],
+    resources: dict,
+    resolved_parameters: dict[str, StackParameter],
+    # TODO: remove resolved conditions somehow
+    resolved_conditions: dict[str, bool],
+    reverse: bool = False,
+) -> list[ChangeConfig]:
+    """
+    Given a list of changes, a dictionary of resources and a dictionary of resolved conditions, topologically sort the
+    changes based on inter-resource dependencies (e.g. usages of intrinsic functions).
+    """
+    ordered_resources = order_resources(
+        resources=resources,
+        resolved_parameters=resolved_parameters,
+        resolved_conditions=resolved_conditions,
+        reverse=reverse,
+    )
+    sorted_changes = []
+    for logical_resource_id in ordered_resources.keys():
+        for change in given_changes:
+            if change["ResourceChange"]["LogicalResourceId"] == logical_resource_id:
+                sorted_changes.append(change)
+                break
+    assert len(sorted_changes) > 0
+    if reverse:
+        sorted_changes = sorted_changes[::-1]
+    return sorted_changes