fix: Stop components in a deterministic order

internal/dag/dag.go

@@ -228,3 +228,7 @@ func (g *Graph) Clone() *Graph {
 	}
 	return newGraph
 }
+
+func (g *Graph) Len() int {
+	return len(g.nodes)
+}

internal/dag/weak.go

@@ -0,0 +1,73 @@
+package dag
+
+type FilterFunc func(g *Graph, n Node) bool
+
+// FilterAllFunc includes every node.
+func FilterAllFunc(_ *Graph, _ Node) bool {
+	return true
+}
+
+// FilterLeavesFunc includes only leaves.
+func FilterLeavesFunc(g *Graph, n Node) bool {
+	return len(g.outEdges[n]) == 0
+}
+
+// FilterRootsFunc includes only roots.
+func FilterRootsFunc(g *Graph, n Node) bool {
+	return len(g.inEdges[n]) == 0
+}
+
+// WeaklyConnectedComponents returns the graph split into weakly connected
+// components. Two nodes are in the same component if there is a path between them in either direction.
+// Each node appears in exactly one component. The graph is unchanged.
+func WeaklyConnectedComponents(g *Graph, f FilterFunc) [][]Node {
+	visited := make(nodeSet)
+	var components [][]Node
+
+	for _, n := range g.Nodes() {
+		if visited.Has(n) {
+			continue
+		}
+
+		var (
+			queue     = []Node{n}
+			component = make([]Node, 0)
+		)
+
+		visited.Add(n)
+		for len(queue) > 0 {
+			curr := queue[0]
+			queue = queue[1:]
+
+			if f(g, curr) {
+				component = append(component, curr)
+			}
+
+			for _, neighbor := range neighbors(g, curr) {
+				if visited.Has(neighbor) {
+					continue
+				}
+				visited.Add(neighbor)
+				queue = append(queue, neighbor)
+			}
+		}
+
+		if len(component) > 0 {
+			components = append(components, component)
+		}
+	}
+
+	return components
+}
+
+// neighbors returns all nodes adjacent to n ignoring edge direction.
+func neighbors(g *Graph, n Node) []Node {
+	out := make([]Node, 0, len(g.outEdges[n])+len(g.inEdges[n]))
+	for neighbor := range g.outEdges[n] {
+		out = append(out, neighbor)
+	}
+	for neighbor := range g.inEdges[n] {
+		out = append(out, neighbor)
+	}
+	return out
+}

internal/dag/weak_test.go

@@ -0,0 +1,94 @@
+package dag
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestWeaklyConnectedComponents(t *testing.T) {
+	t.Run("two disjoint graphs", func(t *testing.T) {
+		var g Graph
+		a, b, c, d := stringNode("a"), stringNode("b"), stringNode("c"), stringNode("d")
+		g.Add(a)
+		g.Add(b)
+		g.Add(c)
+		g.Add(d)
+		g.AddEdge(Edge{From: a, To: b})
+		g.AddEdge(Edge{From: c, To: d})
+
+		got := WeaklyConnectedComponents(&g, FilterAllFunc)
+		got = sortComponents(got)
+
+		require.Equal(t, [][]Node{{a, b}, {c, d}}, got)
+	})
+
+	t.Run("single graph", func(t *testing.T) {
+		var g Graph
+		a, b, c := stringNode("a"), stringNode("b"), stringNode("c")
+		g.Add(a)
+		g.Add(b)
+		g.Add(c)
+		g.AddEdge(Edge{From: a, To: b})
+		g.AddEdge(Edge{From: b, To: c})
+
+		got := WeaklyConnectedComponents(&g, FilterAllFunc)
+		got = sortComponents(got)
+		require.Equal(t, [][]Node{{a, b, c}}, got)
+	})
+
+	t.Run("isolated graphs", func(t *testing.T) {
+		var g Graph
+		a, b, c := stringNode("a"), stringNode("b"), stringNode("c")
+		g.Add(a)
+		g.Add(b)
+		g.Add(c)
+
+		got := WeaklyConnectedComponents(&g, FilterAllFunc)
+		got = sortComponents(got)
+		require.Equal(t, [][]Node{{a}, {b}, {c}}, got)
+	})
+
+	t.Run("only leaves", func(t *testing.T) {
+		var g Graph
+		a, b, c, d := stringNode("a"), stringNode("b"), stringNode("c"), stringNode("d")
+		g.Add(a)
+		g.Add(b)
+		g.Add(c)
+		g.Add(d)
+		g.AddEdge(Edge{From: a, To: b})
+		g.AddEdge(Edge{From: c, To: d})
+
+		got := WeaklyConnectedComponents(&g, FilterLeavesFunc)
+		got = sortComponents(got)
+		require.Equal(t, [][]Node{{b}, {d}}, got)
+	})
+
+	t.Run("only roots", func(t *testing.T) {
+		var g Graph
+		a, b, c, d := stringNode("a"), stringNode("b"), stringNode("c"), stringNode("d")
+		g.Add(a)
+		g.Add(b)
+		g.Add(c)
+		g.Add(d)
+		g.AddEdge(Edge{From: a, To: b})
+		g.AddEdge(Edge{From: c, To: d})
+
+		got := WeaklyConnectedComponents(&g, FilterRootsFunc)
+		got = sortComponents(got)
+		require.Equal(t, [][]Node{{a}, {c}}, got)
+	})
+}
+
+func sortComponents(components [][]Node) [][]Node {
+	for _, c := range components {
+		sort.Slice(c, func(i, j int) bool {
+			return c[i].NodeID() < c[j].NodeID()
+		})
+	}
+	sort.Slice(components, func(i, j int) bool {
+		return components[i][0].NodeID() < components[j][0].NodeID()
+	})
+	return components
+}

internal/runtime/alloy.go

@@ -265,9 +265,11 @@ func newController(o controllerOptions) (*Runtime, error) {
 // Run starts the Alloy controller, blocking until the provided context is
 // canceled. Run must only be called once.
 func (f *Runtime) Run(ctx context.Context) {
-	defer func() { _ = f.sched.Close() }()
-	defer f.loader.Cleanup(!f.opts.IsModule)
-	defer level.Debug(f.log).Log("msg", "Alloy controller exiting")
+	defer func() {
+		level.Debug(f.log).Log("msg", "Alloy controller exiting")
+		f.loader.Cleanup(!f.opts.IsModule)
+		f.sched.Stop()
+	}()
 
 	level.Debug(f.log).Log("msg", "Running alloy controller")
 
@@ -284,34 +286,9 @@ func (f *Runtime) Run(ctx context.Context) {
 			f.loader.EvaluateDependants(ctx, all)
 		case <-f.loadFinished:
 			level.Info(f.log).Log("msg", "scheduling loaded components and services")
-
-			var (
-				components = f.loader.Components()
-				services   = f.loader.Services()
-				imports    = f.loader.Imports()
-
-				runnables = make([]controller.RunnableNode, 0, len(components)+len(services)+len(imports))
-			)
-			for _, c := range components {
-				runnables = append(runnables, c)
-			}
-
-			for _, i := range imports {
-				runnables = append(runnables, i)
-			}
-
-			// Only the root controller should run services, since modules share the
-			// same service instance as the root.
-			if !f.opts.IsModule {
-				for _, svc := range services {
-					runnables = append(runnables, svc)
-				}
-			}
-
-			if err := f.sched.Synchronize(runnables); err != nil {
+			if err := f.sched.Synchronize(f.loader.Graph()); err != nil {
 				level.Error(f.log).Log("msg", "failed to load components and services", "err", err)
 			}
-
 			f.loadComplete.Store(true)
 		}
 	}