[fix] Recursion caused by cycle step

.github/workflows/release.yml

@@ -7,7 +7,7 @@ on:
     branches:
       - main
       - test
-      - 112
+      - 112-cycle-patch
       - dev-release*
 
 defaults:

pkg/apis/numaflow/v1alpha1/pipeline_types.go

@@ -161,18 +161,23 @@ func (p Pipeline) GetAllBuckets() []string {
 
 // GetDownstreamEdges returns all the downstream edges of a vertex
 func (p Pipeline) GetDownstreamEdges(vertexName string) []Edge {
-	var f func(vertexName string, edges *[]Edge)
-	f = func(vertexName string, edges *[]Edge) {
-		for _, b := range p.ListAllEdges() {
-			if b.From == vertexName {
-				*edges = append(*edges, b)
-				f(b.To, edges)
-			}
+    var f func(vertexName string, edges *[]Edge, visited *map[string]bool)
+    f = func(vertexName string, edges *[]Edge, visited *map[string]bool) {
+        if (*visited)[vertexName] {
+			return
 		}
-	}
-	result := []Edge{}
-	f(vertexName, &result)
-	return result
+        for _, b := range p.ListAllEdges() {
+            if b.From == vertexName {
+                *edges = append(*edges, b)
+                (*visited)[vertexName] = true
+				f(b.To, edges, visited)
+            }
+        }
+    }
+    result := []Edge{}
+    visited := make(map[string]bool)
+    f(vertexName, &result, &visited)
+    return result
 }
 
 // HasSideInputs returns if the pipeline has side inputs.

pkg/apis/numaflow/v1alpha1/pipeline_types_test.go

@@ -301,18 +301,26 @@ func Test_GetDownstreamEdges(t *testing.T) {
 			Edges: []Edge{
 				{From: "input", To: "p1"},
 				{From: "p1", To: "p11"},
+				{From: "p1", To: "p1"},
 				{From: "p1", To: "p2"},
 				{From: "p2", To: "output"},
 			},
 		},
 	}
 	edges := pl.GetDownstreamEdges("input")
-	assert.Equal(t, 4, len(edges))
+	assert.Equal(t, 5, len(edges))
 	assert.Equal(t, edges, pl.ListAllEdges())
-	assert.Equal(t, edges[2], Edge{From: "p1", To: "p2"})
+	assert.Equal(t, edges[2], Edge{From: "p1", To: "p1"})
+	assert.Equal(t, edges[3], Edge{From: "p1", To: "p2"})
 
 	edges = pl.GetDownstreamEdges("p1")
-	assert.Equal(t, 3, len(edges))
+	assert.Equal(t, 4, len(edges))
+
+	edges = pl.GetDownstreamEdges("p2")
+	assert.Equal(t, 1, len(edges))
+
+	edges = pl.GetDownstreamEdges("p11")
+	assert.Equal(t, 0, len(edges))
 
 	edges = pl.GetDownstreamEdges("output")
 	assert.Equal(t, 0, len(edges))