reserve only from checklists with successful primary stages

commands/reserve_task.go

@@ -95,22 +95,33 @@ func findNextTaskToRun(cocoon *db.Cocoon, agent *db.Agent) (*db.TaskEntity, *db.
 
 	for ci := len(checklists) - 1; ci >= 0; ci-- {
 		checklist := checklists[ci]
-		tasks, err := cocoon.QueryTasks(checklist.Key)
+		stages, err := cocoon.QueryTasksGroupedByStage(checklist.Key)
+
+		if !allPrimaryStagesSuccessful(stages) {
+			continue
+		}
 
 		if err != nil {
 			return nil, nil, err
 		}
 
-		for _, taskEntity := range tasks {
-			task := taskEntity.Task
-
-			if len(task.RequiredCapabilities) == 0 {
-				log.Errorf(cocoon.Ctx, "Task %v has no required capabilities", task.Name)
+		for _, stage := range stages {
+			if stage.IsPrimary() {
+				// Primary stages are run by Travis and Chromebots. We do not reserve
+				// these tasks for agents.
 				continue
 			}
+			for _, taskEntity := range stage.Tasks {
+				task := taskEntity.Task
+
+				if len(task.RequiredCapabilities) == 0 {
+					log.Errorf(cocoon.Ctx, "Task %v has no required capabilities", task.Name)
+					continue
+				}
 
-			if task.Status == "New" && agent.CapableOfPerforming(task) {
-				return taskEntity, checklist, nil
+				if task.Status == "New" && agent.CapableOfPerforming(task) {
+					return taskEntity, checklist, nil
+				}
 			}
 		}
 	}
@@ -153,3 +164,14 @@ func atomicallyReserveTask(cocoon *db.Cocoon, taskKey *datastore.Key, agent *db.
 
 	return taskEntity, nil
 }
+
+func allPrimaryStagesSuccessful(stages []*db.Stage) bool {
+	isSuccessfulPrimaryOrAnySecondary := func(istage interface{}) bool {
+		stage := istage.(*db.Stage)
+		if stage.IsPrimary() {
+			return stage.Status == db.TaskSucceeded
+		}
+		return true
+	}
+	return db.Every(len(stages), func(i int) interface{} { return stages[i] }, isSuccessfulPrimaryOrAnySecondary)
+}

db/collection_utils.go

@@ -0,0 +1,37 @@
+// Copyright (c) 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package db
+
+// Every is like Dart's Iterable.every.
+func Every(len int, getter func(i int) interface{}, predicate func(element interface{}) bool) bool {
+	for i := 0; i < len; i++ {
+		if !predicate(getter(i)) {
+			return false
+		}
+	}
+	return true
+}
+
+// Any is like Dart's Iterable.any.
+func Any(len int, getter func(i int) interface{}, predicate func(element interface{}) bool) bool {
+	for i := 0; i < len; i++ {
+		if predicate(getter(i)) {
+			return true
+		}
+	}
+	return false
+}
+
+// Where is like Dart's Iterable.where.
+func Where(len int, getter func(i int) interface{}, predicate func(element interface{}) bool) []interface{} {
+	var results []interface{}
+	for i := 0; i < len; i++ {
+		element := getter(i)
+		if predicate(element) {
+			results = append(results, element)
+		}
+	}
+	return results
+}

db/db.go

@@ -239,13 +239,60 @@ func (c *Cocoon) QueryTasksGroupedByStage(checklistKey *datastore.Key) ([]*Stage
 	stages := make([]*Stage, len(stageMap))
 	i := 0
 	for _, stage := range stageMap {
+		stage.Status = computeStageStatus(stage)
 		stages[i] = stage
 		i++
 	}
 	sort.Sort(byPrecedence(stages))
 	return stages, nil
 }
 
+// See docs on the Stage.Status property.
+func computeStageStatus(stage *Stage) TaskStatus {
+	len := len(stage.Tasks)
+
+	getter := func(i int) interface{} {
+		return stage.Tasks[i]
+	}
+
+	isSucceeded := func(task interface{}) bool {
+		return task.(*TaskEntity).Task.Status == TaskSucceeded
+	}
+	if Every(len, getter, isSucceeded) {
+		return TaskSucceeded
+	}
+
+	isFailed := func(task interface{}) bool {
+		return task.(*TaskEntity).Task.Status == TaskFailed
+	}
+	if Any(len, getter, isFailed) {
+		return TaskFailed
+	}
+
+	isInProgress := func(task interface{}) bool {
+		return task.(*TaskEntity).Task.Status == TaskInProgress
+	}
+	isInProgressOrNew := func(task interface{}) bool {
+		return task.(*TaskEntity).Task.Status == TaskInProgress || task.(*TaskEntity).Task.Status == TaskNew
+	}
+	if Any(len, getter, isInProgress) && Every(len, getter, isInProgressOrNew) {
+		return TaskInProgress
+	}
+
+	prevStatus := TaskNoStatus
+	isSameAsPrevious := func(task interface{}) bool {
+		status := task.(*TaskEntity).Task.Status
+		result := prevStatus == TaskNoStatus || prevStatus == status
+		prevStatus = status
+		return result
+	}
+	if Every(len, getter, isSameAsPrevious) && prevStatus != TaskNoStatus {
+		return prevStatus
+	}
+
+	return TaskFailed
+}
+
 type byPrecedence []*Stage
 
 func (stages byPrecedence) Len() int      { return len(stages) }
@@ -429,3 +476,13 @@ func (agent *Agent) CapableOfPerforming(task *Task) bool {
 
 	return true
 }
+
+// IsPrimary returns whether stage is primary or not.
+//
+// There are two categories of stages: primary and secondary. Tasks in the
+// secondary stages are only performed if all tasks in the primary stages are
+// successful.
+func (stage *Stage) IsPrimary() bool {
+	name := stage.Name
+	return name == "travis" || name == "chromebot"
+}

db/schema.go

@@ -39,8 +39,16 @@ type Checklist struct {
 // tasks into a pipeline, where tasks in some stages only run _after_ a previous
 // stage is successful.
 type Stage struct {
-	Name  string
-	Tasks []*TaskEntity
+	Name string
+	// Aggregated status of the stage, computed as follows:
+	//
+	// - TaskSucceeded if all tasks in this stage succeeded
+	// - TaskFailed if at least one task in this stage failed
+	// - TaskInProgress if at least one task is in progress and others are New
+	// - Same as Task.Status if all tasks have the same status
+	// - TaskFailed otherwise
+	Status TaskStatus
+	Tasks  []*TaskEntity
 }
 
 // TaskEntity contains storage data on a Task.
@@ -69,6 +77,11 @@ type Task struct {
 // TaskStatus indicates the status of a task.
 type TaskStatus string
 
+// TaskNoStatus is the zero value, meaning no status value. It is not a valid
+// status value and should only be used as a temporary variable value in
+// algorithms that need it.
+const TaskNoStatus = TaskStatus("")
+
 // TaskNew indicates that the task was created but not acted upon.
 const TaskNew = TaskStatus("New")