feat: cancel successor steps if previous failed (#68)

Author: gabyx

pkg/dag/node.go

@@ -100,6 +100,17 @@ func (n *TargetNode) Status() ExecStatus {
 	return ExecStatusSuccess
 }
 
+// StatusAnyFailed determines if any runner in the target has failed.
+func (n *TargetNode) StatusAnyFailed() bool {
+	for _, r := range n.Execution.Runners {
+		if r.Status == ExecStatusFailed {
+			return true
+		}
+	}
+
+	return false
+}
+
 // PropagateExecStatus propagates the execution status forward.
 func (n *TargetNode) PropagateExecStatus() {
 	for _, f := range n.Forward {

pkg/dag/run-concurrent.go

@@ -161,14 +161,24 @@ func addRunnerTasks(
 				runner.RunnerID,
 			}
 
+			// Always on finish propagate exec status.
+			defer func() { node.PropagateExecStatus() }()
+
 			if node.Execution.Cancel {
 				log.Debugf(
 					"Runner '%v' for target '%v' is cancelled by dependency.",
 					runnerIdx,
 					node.Target.ID,
 				)
 
-				node.PropagateExecStatus()
+				return
+			} else if node.StatusAnyFailed() {
+				log.Debugf(
+					"Target '%v' is failed already. Skip runner '%v', step: '%v'.",
+					node.Target.ID,
+					runner.RunnerID,
+					stepIdx,
+				)
 
 				return
 			}
@@ -191,8 +201,6 @@ func addRunnerTasks(
 				} else {
 					status.Status = ExecStatusSuccess
 				}
-
-				node.PropagateExecStatus()
 			}()
 
 			err = ExecuteRunner(

pkg/dag/run.go

@@ -178,13 +178,21 @@ func executeRunners(
 			rD.inst.RunnerID,
 		}
 
-		if rD.node.Execution.Cancel {
+		switch {
+		case rD.node.Execution.Cancel:
 			log.Debugf(
 				"Target '%v' is cancelled by prev. target. Skip runner '%v'",
+				rD.node.Target.ID,
 				rD.inst.RunnerID,
+			)
+		case rD.node.StatusAnyFailed():
+			log.Debugf(
+				"Target '%v' is failed already. Skip runner '%v', step: '%v'.",
 				rD.node.Target.ID,
+				rD.inst.RunnerID,
+				rD.step.Index,
 			)
-		} else {
+		default:
 			log.Info("Starting runner.", "runner", rD.inst.RunnerID, "target", rD.targetID)
 
 			e := ExecuteRunner(