On Pipeline creation, verify order of providedBy constraints

Tasks should only rely on outputs of tasks that execute before them,
they shouldn't be able to rely on outputs of future tasks. This also
means we no longer need the `canRunTask` function: since Tasks currently
execute linearly (see tektoncd#168), the only way `canRunTask` could return
false would be if `Tasks` in `providedBy` actually executed _after_ the
one currently being evaluated, which would result in an unrunnable and
invalid Pipeline. Now we will catch this on Pipeline creation.

Part of tektoncd#320

pkg/apis/pipeline/v1alpha1/pipeline_validation.go

@@ -48,12 +48,24 @@ func (ps *PipelineSpec) Validate() *apis.FieldError {
 		taskNames[t.Name] = struct{}{}
 	}
 
-	// providedBy should match other tasks.
-	for _, t := range ps.Tasks {
+	// providedBy should match future tasks
+	// TODO(#168) when pipelines don't just execute linearly this will need to be more sophisticated
+	for i, t := range ps.Tasks {
 		if t.Resources != nil {
 			for _, rd := range t.Resources.Inputs {
 				for _, pb := range rd.ProvidedBy {
-					if _, ok := taskNames[pb]; !ok {
+					if i == 0 {
+						// First Task can't depend on anything before it (b/c there is nothing)
+						return apis.ErrInvalidKeyName(pb, fmt.Sprintf("spec.tasks.resources.%s", pb))
+					}
+					found := false
+					// Look for previous Task that satisfies constraint
+					for j := i - 1; j >= 0; j-- {
+						if ps.Tasks[j].Name == pb {
+							found = true
+						}
+					}
+					if !found {
 						return apis.ErrInvalidKeyName(pb, fmt.Sprintf("spec.tasks.resources.%s", pb))
 					}
 				}

pkg/apis/pipeline/v1alpha1/pipeline_validation_test.go

@@ -40,7 +40,7 @@ func TestPipelineSpec_Validate_Error(t *testing.T) {
 			},
 		},
 		{
-			name: "invalid constraint tasks",
+			name: "providedby task doesnt exist",
 			fields: fields{
 				Tasks: []PipelineTask{{
 					Name: "foo",
@@ -52,6 +52,21 @@ func TestPipelineSpec_Validate_Error(t *testing.T) {
 				}},
 			},
 		},
+		{
+			name: "providedby task is afterward",
+			fields: fields{
+				Tasks: []PipelineTask{{
+					Name: "foo",
+					Resources: &PipelineTaskResources{
+						Inputs: []PipelineTaskInputResource{{
+							ProvidedBy: []string{"bar"},
+						}},
+					},
+				}, {
+					Name: "bar",
+				}},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -89,14 +104,14 @@ func TestPipelineSpec_Validate_Valid(t *testing.T) {
 			name: "valid constraint tasks",
 			fields: fields{
 				Tasks: []PipelineTask{{
+					Name: "bar",
+				}, {
 					Name: "foo",
 					Resources: &PipelineTaskResources{
 						Inputs: []PipelineTaskInputResource{{
 							ProvidedBy: []string{"bar"},
 						}},
 					},
-				}, {
-					Name: "bar",
 				}},
 			},
 		},

pkg/reconciler/v1alpha1/pipelinerun/resources/pipelinerunresolution.go

@@ -60,7 +60,7 @@ func GetNextTask(prName string, state []*ResolvedPipelineRunTask, logger *zap.Su
 				logger.Infof("TaskRun %s is still running so we shouldn't start more for PipelineRun %s", prtr.TaskRunName, prName)
 				return nil
 			}
-		} else if canTaskRun(prtr.PipelineTask, state) {
+		} else {
 			logger.Infof("TaskRun %s should be started for PipelineRun %s", prtr.TaskRunName, prName)
 			return prtr
 		}
@@ -69,37 +69,6 @@ func GetNextTask(prName string, state []*ResolvedPipelineRunTask, logger *zap.Su
 	return nil
 }
 
-func canTaskRun(pt *v1alpha1.PipelineTask, state []*ResolvedPipelineRunTask) bool {
-	// Check if Task can run now. Go through all the input constraints
-	if pt.Resources != nil {
-		for _, rd := range pt.Resources.Inputs {
-			if len(rd.ProvidedBy) > 0 {
-				for _, constrainingTaskName := range rd.ProvidedBy {
-					for _, prtr := range state {
-						// the constraining task must have a successful task run to allow this task to run
-						if prtr.PipelineTask.Name == constrainingTaskName {
-							if prtr.TaskRun == nil {
-								return false
-							}
-							c := prtr.TaskRun.Status.GetCondition(duckv1alpha1.ConditionSucceeded)
-							if c == nil {
-								return false
-							}
-							switch c.Status {
-							case corev1.ConditionFalse:
-								return false
-							case corev1.ConditionUnknown:
-								return false
-							}
-						}
-					}
-				}
-			}
-		}
-	}
-	return true
-}
-
 // ResolvedPipelineRunTask contains a Task and its associated TaskRun, if it
 // exists. TaskRun can be nil to represent there being no TaskRun.
 type ResolvedPipelineRunTask struct {