fix cycle detection logic

visited map holds name of the nodes been visited as keys and "true"
as value. This map is being updated with currentName.HashKey on every
iteration which results in keys such as a.d, b.d where a, b, and d
are nodes of a dag. But such keys are not utilized anywhere in the
visit function. Visit function checks existence of the node just
by the name without any string concatenation.

This extra addition in the map is causing severe delay for a graph
with more than >60 nodes.

pkg/reconciler/pipeline/dag/dag.go

@@ -128,26 +128,21 @@ func linkPipelineTasks(prev *Node, next *Node) error {
 	// Check if we are adding cycles.
 	visited := map[string]bool{prev.Task.HashKey(): true, next.Task.HashKey(): true}
 	path := []string{next.Task.HashKey(), prev.Task.HashKey()}
-	if err := visit(next.Task.HashKey(), prev.Prev, path, visited); err != nil {
+	if err := visit(prev.Prev, path, visited); err != nil {
 		return fmt.Errorf("cycle detected: %w", err)
 	}
 	next.Prev = append(next.Prev, prev)
 	prev.Next = append(prev.Next, next)
 	return nil
 }
 
-func visit(currentName string, nodes []*Node, path []string, visited map[string]bool) error {
-	var sb strings.Builder
+func visit(nodes []*Node, path []string, visited map[string]bool) error {
 	for _, n := range nodes {
 		path = append(path, n.Task.HashKey())
 		if _, ok := visited[n.Task.HashKey()]; ok {
 			return errors.New(getVisitedPath(path))
 		}
-		sb.WriteString(currentName)
-		sb.WriteByte('.')
-		sb.WriteString(n.Task.HashKey())
-		visited[sb.String()] = true
-		if err := visit(n.Task.HashKey(), n.Prev, path, visited); err != nil {
+		if err := visit(n.Prev, path, visited); err != nil {
 			return err
 		}
 	}

pkg/reconciler/pipeline/dag/dag_test.go

@@ -669,3 +669,59 @@ func TestBuild_ConditionsParamsFromTaskResults(t *testing.T) {
 	}
 	assertSameDAG(t, expectedDAG, g)
 }
+
+// This test make sure we detect cycle when it exists.
+// This means we "visit" "a" twice, from two different paths.
+// And one of the paths is creating a cycle with "e -> a".
+//   a
+//  / \
+// b   c
+//  \ /
+//   d
+//  / \
+// e   f
+//     |
+//     g
+func TestBuild_InvalidDAGWithCycle(t *testing.T) {
+	a := v1beta1.PipelineTask{Name: "a", RunAfter: []string{"e"}}
+	bDependsOnA := v1beta1.PipelineTask{
+		Name: "b",
+		Resources: &v1beta1.PipelineTaskResources{
+			Inputs: []v1beta1.PipelineTaskInputResource{{From: []string{"a"}}},
+		},
+	}
+	cRunsAfterA := v1beta1.PipelineTask{
+		Name:     "c",
+		RunAfter: []string{"a"},
+	}
+	dDependsOnBAndC := v1beta1.PipelineTask{
+		Name: "f",
+		Resources: &v1beta1.PipelineTaskResources{
+			Inputs: []v1beta1.PipelineTaskInputResource{{From: []string{"b", "c"}}},
+		},
+	}
+	eRunsAfterD := v1beta1.PipelineTask{
+		Name:     "e",
+		RunAfter: []string{"d"},
+	}
+	fRunsAfterD := v1beta1.PipelineTask{
+		Name:     "f",
+		RunAfter: []string{"d"},
+	}
+	gDependsOnF := v1beta1.PipelineTask{
+		Name: "g",
+		Resources: &v1beta1.PipelineTaskResources{
+			Inputs: []v1beta1.PipelineTaskInputResource{{From: []string{"f"}}},
+		},
+	}
+
+	p := &v1beta1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "invalid-dag-with-cycle"},
+		Spec: v1beta1.PipelineSpec{
+			Tasks: v1beta1.PipelineTaskList{a, bDependsOnA, cRunsAfterA, dDependsOnBAndC, eRunsAfterD, fRunsAfterD, gDependsOnF},
+		},
+	}
+	if _, err := dag.Build(v1beta1.PipelineTaskList(p.Spec.Tasks)); err == nil {
+		t.Errorf("expected to see an error for invalid DAG in pipeline %v but had none", p.Spec)
+	}
+}