feat: implement workflow validation to reject duplicate edges and remove redundant edge filtering logic (#808)

Author: hanorik

workflow/validation.go

@@ -29,6 +29,11 @@ var ErrNoStartNode = errors.New("no start node found")
 // ErrNodePointsToStart is returned when a node points to the start node.
 var ErrNodePointsToStart = errors.New("node points to start node")
 
+// ErrDuplicateEdge is returned when an edge set contains two identical edges.
+// Two edges with the same (From, To) are rejected regardless of Route; use
+// MultiRoute to express alternatives to the same target.
+var ErrDuplicateEdge = errors.New("duplicate edge")
+
 // ErrMultipleDefaultRoutes is returned when a node has more than one default route.
 var ErrMultipleDefaultRoutes = errors.New("node has more than one default route")
 
@@ -97,7 +102,10 @@ func validateStartNodeNoIncoming(edges []Edge) error {
 }
 
 // validateWorkflow executes a set of workflow validation checks.
-func validateWorkflow(workflow *Workflow) error {
+func validateWorkflow(workflow *graph) error {
+	if err := validateUniqueEdges(workflow); err != nil {
+		return err
+	}
 	if err := validateDefaultRoute(workflow); err != nil {
 		return err
 	}
@@ -107,9 +115,25 @@ func validateWorkflow(workflow *Workflow) error {
 	return nil
 }
 
+// validateUniqueEdges checks that there are no duplicate edges in the workflow.
+// Two edges with the same (From, To) are rejected regardless of Route; use
+// MultiRoute to express alternatives to the same target.
+func validateUniqueEdges(workflow *graph) error {
+	for node, edges := range workflow.successors {
+		uniqueEdges := make(map[Node]struct{})
+		for _, edge := range edges {
+			if _, ok := uniqueEdges[edge.To]; ok {
+				return fmt.Errorf("%w: from %q to %q", ErrDuplicateEdge, node.Name(), edge.To.Name())
+			}
+			uniqueEdges[edge.To] = struct{}{}
+		}
+	}
+	return nil
+}
+
 // validateDefaultRoute checks that there are no multiple default routes for one node.
-func validateDefaultRoute(workflow *Workflow) error {
-	for node, edges := range workflow.graph.successors {
+func validateDefaultRoute(workflow *graph) error {
+	for node, edges := range workflow.successors {
 		hasDefault := false
 		for _, edge := range edges {
 			if edge.Route == Default && !hasDefault {
@@ -127,7 +151,7 @@ func validateDefaultRoute(workflow *Workflow) error {
 // for cycles where all edges in the cycle have nil routes.
 // Default routes (where Route == Default) are treated as conditional edges
 // and are ignored during unconditional cycle detection.
-func validateCycles(workflow *Workflow) error {
+func validateCycles(workflow *graph) error {
 	visited := make(map[Node]struct{})
 
 	var traverse func(n Node, inStack map[Node]struct{}) error
@@ -143,7 +167,7 @@ func validateCycles(workflow *Workflow) error {
 		inStack[n] = struct{}{}
 		visited[n] = struct{}{}
 
-		for _, edge := range workflow.graph.successors[n] {
+		for _, edge := range workflow.successors[n] {
 			if edge.Route == nil {
 				if err := traverse(edge.To, inStack); err != nil {
 					return err
@@ -155,7 +179,7 @@ func validateCycles(workflow *Workflow) error {
 		return nil
 	}
 
-	for node := range workflow.graph.successors {
+	for node := range workflow.successors {
 		if _, ok := visited[node]; !ok {
 			inStack := make(map[Node]struct{})
 			if err := traverse(node, inStack); err != nil {

workflow/validation_test.go

@@ -155,6 +155,50 @@ func TestStartNodeNoIncomingEdges(t *testing.T) {
 	}
 }
 
+func TestDuplicateEdges(t *testing.T) {
+	nodeA := &dummyNode{name: "A"}
+	nodeB := &dummyNode{name: "B"}
+	tests := []struct {
+		name      string
+		edges     []Edge
+		expectErr bool
+	}{
+		{
+			name:  "no duplicate edges",
+			edges: []Edge{{From: nodeA, To: nodeB}},
+		},
+		{
+			name:      "duplicate edges",
+			edges:     []Edge{{From: nodeA, To: nodeB}, {From: nodeA, To: nodeB}},
+			expectErr: true,
+		},
+		{
+			name:      "duplicate edges with different routes",
+			edges:     []Edge{{From: nodeA, To: nodeB, Route: StringRoute("test1")}, {From: nodeA, To: nodeB, Route: StringRoute("test2")}},
+			expectErr: true,
+		},
+		{
+			name:      "duplicate edges one without route",
+			edges:     []Edge{{From: nodeA, To: nodeB, Route: StringRoute("test1")}, {From: nodeA, To: nodeB}},
+			expectErr: true,
+		},
+		{
+			name:  "empty edges",
+			edges: []Edge{},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if err := validateUniqueEdges(newGraph(tc.edges)); err != nil && !tc.expectErr {
+				t.Errorf("got an error %v, expected none", err)
+			} else if err == nil && tc.expectErr {
+				t.Errorf("expected an error, got none")
+			}
+		})
+	}
+}
+
 func TestDefaultRoute(t *testing.T) {
 	nodeA := &dummyNode{name: "A"}
 	nodeB := &dummyNode{name: "B"}
@@ -177,7 +221,7 @@ func TestDefaultRoute(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			if err := validateDefaultRoute(&Workflow{graph: newGraph(tc.edges)}); !errors.Is(err, tc.expectErr) {
+			if err := validateDefaultRoute(newGraph(tc.edges)); !errors.Is(err, tc.expectErr) {
 				t.Errorf("got %v, expected %v", err, tc.expectErr)
 			}
 		})
@@ -272,9 +316,7 @@ func TestValidateCycles(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			w := &Workflow{graph: newGraph(tc.edges())}
-
-			err := validateCycles(w)
+			err := validateCycles(newGraph(tc.edges()))
 			if tc.expectErr && err == nil {
 				t.Errorf("expected error, got none")
 			} else if !tc.expectErr && err != nil {

workflow/workflow.go

@@ -120,7 +120,11 @@ func New(edges []Edge) (*Workflow, error) {
 	if err := validateNodes(edges); err != nil {
 		return nil, err
 	}
-	return &Workflow{graph: newGraph(edges)}, nil
+	graph := newGraph(edges)
+	if err := validateWorkflow(graph); err != nil {
+		return nil, err
+	}
+	return &Workflow{graph: graph}, nil
 }
 
 // Run drives the workflow to completion (or to a graceful pause in

workflow/workflow_test.go

@@ -426,19 +426,15 @@ func TestWorkflowRouting(t *testing.T) {
 			expectedExec: nil,
 		},
 		{
-			name:        "duplicate edges to same node",
-			startRoutes: []string{"branchA"},
+			name:        "MultiRoute with multiple matching routes",
+			startRoutes: []string{"branchA", "branchB"},
 			edges: func(x *CustomRouteNode, a, b *FunctionNode, c *CustomRouteNode, d *FunctionNode) []Edge {
 				return []Edge{
 					{From: Start, To: x},
-					{From: x, To: a},
-					{From: x, To: a, Route: StringRoute("branchA")},
-					{From: x, To: b},
-					{From: x, To: c},
-					{From: c, To: d},
+					{From: x, To: a, Route: MultiRoute[string]{"branchA", "branchB"}},
 				}
 			},
-			expectedExec: []string{"A", "B", "C", "D"},
+			expectedExec: []string{"A"},
 		},
 	}