Include transitive children in dependency list for deletes

CHANGELOG_PENDING.md

@@ -11,3 +11,6 @@
 - [sdk/dotnet] - Fix an exception when passing an unknown `Output` to
   the `DependsOn` resource option.
   [#7762](https://github.com/pulumi/pulumi/pull/7762)
+
+- [engine] Include transitive children in dependency list for deletes.
+  [#7788](https://github.com/pulumi/pulumi/pull/7788)

pkg/resource/graph/dependency_graph.go

@@ -10,8 +10,9 @@ import (
 
 // DependencyGraph represents a dependency graph encoded within a resource snapshot.
 type DependencyGraph struct {
-	index     map[*resource.State]int // A mapping of resource pointers to indexes within the snapshot
-	resources []*resource.State       // The list of resources, obtained from the snapshot
+	index      map[*resource.State]int // A mapping of resource pointers to indexes within the snapshot
+	resources  []*resource.State       // The list of resources, obtained from the snapshot
+	childrenOf map[resource.URN][]int  // Pre-computed map of transitive children for each resource
 }
 
 // DependingOn returns a slice containing all resources that directly or indirectly
@@ -92,7 +93,25 @@ func (dg *DependencyGraph) DependenciesOf(res *resource.State) ResourceSet {
 	contract.Assert(ok)
 	for i := cursorIndex - 1; i >= 0; i-- {
 		candidate := dg.resources[i]
-		if dependentUrns[candidate.URN] || candidate.URN == res.Parent {
+		// Include all resources that are dependencies of the resource
+		if dependentUrns[candidate.URN] {
+			set[candidate] = true
+			// If the dependency is a component, all transitive children of the dependency that are before this
+			// resource in the topological sort are also implicitly dependencies. This is necessary because for remote
+			// components, the dependencies will not include the transitive set of children directly, but will include
+			// the parent component. We must walk that component's children here to ensure they are treated as
+			// dependencies. Transitive children of the dependency that are after the resource in the topological sort
+			// are not included as this could lead to cycles in the dependency order.
+			if !candidate.Custom {
+				for _, transitiveCandidateIndex := range dg.childrenOf[candidate.URN] {
+					if transitiveCandidateIndex < cursorIndex {
+						set[dg.resources[transitiveCandidateIndex]] = true
+					}
+				}
+			}
+		}
+		// Include the resource's parent, as the resource depends on it's parent existing.
+		if candidate.URN == res.Parent {
 			set[candidate] = true
 		}
 	}
@@ -101,12 +120,22 @@ func (dg *DependencyGraph) DependenciesOf(res *resource.State) ResourceSet {
 }
 
 // NewDependencyGraph creates a new DependencyGraph from a list of resources.
-// The resources should be in topological order with respect to their dependencies.
+// The resources should be in topological order with respect to their dependencies, including
+// parents appearing before children.
 func NewDependencyGraph(resources []*resource.State) *DependencyGraph {
 	index := make(map[*resource.State]int)
+	childrenOf := make(map[resource.URN][]int)
+
+	urnIndex := make(map[resource.URN]int)
 	for idx, res := range resources {
 		index[res] = idx
+		urnIndex[res.URN] = idx
+		parent := res.Parent
+		for parent != "" {
+			childrenOf[parent] = append(childrenOf[parent], idx)
+			parent = resources[urnIndex[parent]].Parent
+		}
 	}
 
-	return &DependencyGraph{index, resources}
+	return &DependencyGraph{index, resources, childrenOf}
 }

pkg/resource/graph/dependency_graph_test.go

@@ -173,3 +173,59 @@ func TestDependenciesOf(t *testing.T) {
 	assert.False(t, dDepends[b])
 	assert.False(t, dDepends[c])
 }
+
+func TestDependenciesOfRemoteComponents(t *testing.T) {
+	aws := NewProviderResource("aws", "default", "0")
+	xyz := NewProviderResource("xyz", "default", "0")
+	first := NewResource("first", xyz)
+	firstNested := NewResource("firstNested", xyz)
+	firstNested.Parent = first.URN
+	sg := NewResource("sg", aws)
+	sg.Parent = firstNested.URN
+	second := NewResource("second", xyz)
+	rule := NewResource("rule", aws, first.URN)
+	rule.Parent = second.URN
+
+	dg := NewDependencyGraph([]*resource.State{
+		aws,
+		xyz,
+		first,
+		firstNested,
+		sg,
+		second,
+		rule,
+	})
+
+	ruleDepends := dg.DependenciesOf(rule)
+	assert.True(t, ruleDepends[first], "direct dependency")
+	assert.True(t, ruleDepends[firstNested], "child of dependency")
+	assert.True(t, ruleDepends[sg], "transitive child of dependency")
+	assert.True(t, ruleDepends[second], "parent")
+	assert.True(t, ruleDepends[aws], "provider")
+	assert.False(t, ruleDepends[xyz], "unrelated")
+}
+
+func TestDependenciesOfRemoteComponentsNoCycle(t *testing.T) {
+	aws := NewProviderResource("aws", "default", "0")
+	parent := NewResource("parent", aws)
+	r := NewResource("r", aws, parent.URN)
+	child := NewResource("child", aws, r.URN)
+	child.Parent = parent.URN
+
+	dg := NewDependencyGraph([]*resource.State{
+		aws,
+		parent,
+		r,
+		child,
+	})
+
+	childDependencies := dg.DependenciesOf(child)
+	assert.True(t, childDependencies[aws])
+	assert.True(t, childDependencies[parent])
+	assert.True(t, childDependencies[r])
+
+	rDependencies := dg.DependenciesOf(r)
+	assert.True(t, rDependencies[aws])
+	assert.True(t, rDependencies[parent])
+	assert.False(t, rDependencies[child])
+}