core: dag errors should cascade to all descendents

dag/dag.go

@@ -195,7 +195,7 @@ func (g *AcyclicGraph) Walk(cb WalkFunc) error {
 
 		// Start the goroutine to wait for our dependencies
 		readyCh := make(chan bool)
-		go func(deps []Vertex, chs []<-chan struct{}, readyCh chan<- bool) {
+		go func(v Vertex, deps []Vertex, chs []<-chan struct{}, readyCh chan<- bool) {
 			// First wait for all the dependencies
 			for _, ch := range chs {
 				<-ch
@@ -206,13 +206,14 @@ func (g *AcyclicGraph) Walk(cb WalkFunc) error {
 			defer errLock.Unlock()
 			for _, dep := range deps {
 				if errMap[dep] {
+					errMap[v] = true
 					readyCh <- false
 					return
 				}
 			}
 
 			readyCh <- true
-		}(deps, depChs, readyCh)
+		}(v, deps, depChs, readyCh)
 
 		// Start the goroutine that executes
 		go func(v Vertex, doneCh chan<- struct{}, readyCh <-chan bool) {

dag/dag_test.go

@@ -226,8 +226,10 @@ func TestAcyclicGraphWalk_error(t *testing.T) {
 	g.Add(1)
 	g.Add(2)
 	g.Add(3)
+	g.Add(4)
+	g.Connect(BasicEdge(4, 3))
 	g.Connect(BasicEdge(3, 2))
-	g.Connect(BasicEdge(3, 1))
+	g.Connect(BasicEdge(2, 1))
 
 	var visits []Vertex
 	var lock sync.Mutex