func (db *SQLDB) ResolvingTask(ctx context.Context, logger lager.Logger, taskGuid string) (*models.Task, *models.Task, error) {
	logger = logger.Session("db-resolving-task", lager.Data{"task_guid": taskGuid})
	logger.Info("starting")
	defer logger.Info("complete")
	var beforeTask models.Task
	var afterTask *models.Task
	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		var err error
		afterTask, err = db.fetchTaskForUpdate(ctx, logger, taskGuid, tx)
		if err != nil {
			logger.Error("failed-locking-task", err)
			return err
		}
		beforeTask = *afterTask
		if err = afterTask.ValidateTransitionTo(models.Task_Resolving); err != nil {
			logger.Error("invalid-state-transition", err)
			return err
		}
		now := db.clock.Now().UnixNano()
		_, err = db.update(ctx, logger, tx, tasksTable,
			helpers.SQLAttributes{
				"state":      models.Task_Resolving,
				"updated_at": now,
			},
			"guid = ?", taskGuid,
		)
		if err != nil {
			logger.Error("failed-updating-tasks", err)
			return err
		}
		afterTask.State = models.Task_Resolving
		afterTask.UpdatedAt = now
		return nil
	})
	return &beforeTask, afterTask, err
}
The stager calls this when it wants to claim a completed task.  This ensures that only one
stager ever attempts to handle a completed task
package sqldb
import (
	"context"
	"database/sql"
	"strings"
	"code.cloudfoundry.org/bbs/db/sqldb/helpers"
	"code.cloudfoundry.org/bbs/models"
	"code.cloudfoundry.org/lager"
)
func (db *SQLDB) DesireTask(ctx context.Context, logger lager.Logger, taskDef *models.TaskDefinition, taskGuid, domain string) (*models.Task, error) {
	logger = logger.Session("db-desire-task", lager.Data{"task_guid": taskGuid})
	logger.Info("starting")
	defer logger.Info("complete")
	taskDefData, err := db.serializeModel(logger, taskDef)
	if err != nil {
		logger.Error("failed-serializing-task-definition", err)
		return nil, err
	}
	now := db.clock.Now().UnixNano()
	err = db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		_, err = db.insert(ctx, logger, tx, tasksTable,
			helpers.SQLAttributes{
				"guid":               taskGuid,
				"domain":             domain,
				"created_at":         now,
				"updated_at":         now,
				"first_completed_at": 0,
				"state":              models.Task_Pending,
				"task_definition":    taskDefData,
			},
		)
		return err
	})
	if err != nil {
		logger.Error("failed-inserting-task", err)
		return nil, err
	}
	return &models.Task{
		TaskDefinition:   taskDef,
		TaskGuid:         taskGuid,
		Domain:           domain,
		CreatedAt:        now,
		UpdatedAt:        now,
		FirstCompletedAt: 0,
		State:            models.Task_Pending,
	}, nil
}
func (db *SQLDB) Tasks(ctx context.Context, logger lager.Logger, filter models.TaskFilter) ([]*models.Task, error) {
	logger = logger.Session("db-tasks", lager.Data{"filter": filter})
	logger.Debug("starting")
	defer logger.Debug("complete")
	wheres := []string{}
	values := []interface{}{}
	if filter.Domain != "" {
		wheres = append(wheres, "domain = ?")
		values = append(values, filter.Domain)
	}
	if filter.CellID != "" {
		wheres = append(wheres, "cell_id = ?")
		values = append(values, filter.CellID)
	}
	results := []*models.Task{}
	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		rows, err := db.all(ctx, logger, tx, tasksTable,
			taskColumns, helpers.NoLockRow,
			strings.Join(wheres, " AND "), values...,
		)
		if err != nil {
			logger.Error("failed-query", err)
			return err
		}
		defer rows.Close()
		results, _, _, err = db.fetchTasks(ctx, logger, rows, tx, true)
		if err != nil {
			logger.Error("failed-fetch", err)
			return err
		}
		return nil
	})
	return results, err
}
func (db *SQLDB) TaskByGuid(ctx context.Context, logger lager.Logger, taskGuid string) (*models.Task, error) {
	logger = logger.Session("db-task-by-guid", lager.Data{"task_guid": taskGuid})
	logger.Debug("starting")
	defer logger.Debug("complete")
	var task *models.Task
	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		var err error
		row := db.one(ctx, logger, tx, tasksTable,
			taskColumns, helpers.NoLockRow,
			"guid = ?", taskGuid,
		)
		task, err = db.fetchTask(ctx, logger, row, tx)
		return err
	})
	return task, err
}
func (db *SQLDB) StartTask(ctx context.Context, logger lager.Logger, taskGuid, cellId string) (*models.Task, *models.Task, bool, error) {
	logger = logger.Session("db-start-task", lager.Data{"task_guid": taskGuid, "cell_id": cellId})
	logger.Info("starting")
	defer logger.Info("complete")
	var started bool
	var beforeTask models.Task
	var afterTask *models.Task
	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		var err error
		afterTask, err = db.fetchTaskForUpdate(ctx, logger, taskGuid, tx)
		if err != nil {
			logger.Error("failed-locking-task", err)
			return err
		}
		beforeTask = *afterTask
		if afterTask.State == models.Task_Running && afterTask.CellId == cellId {
			logger.Debug("task-already-running-on-cell")
			return nil
		}
		if err = afterTask.ValidateTransitionTo(models.Task_Running); err != nil {
			logger.Error("failed-to-transition-task-to-running", err)
			return err
		}
		now := db.clock.Now().UnixNano()
		_, err = db.update(ctx, logger, tx, tasksTable,
			helpers.SQLAttributes{
				"state":      models.Task_Running,
				"updated_at": now,
				"cell_id":    cellId,
			},
			"guid = ?", taskGuid,
		)
		if err != nil {
			return err
		}
		afterTask.State = models.Task_Running
		afterTask.UpdatedAt = now
		afterTask.CellId = cellId
		started = true
		return nil
	})
	return &beforeTask, afterTask, started, err
}
func (db *SQLDB) CancelTask(ctx context.Context, logger lager.Logger, taskGuid string) (*models.Task, *models.Task, string, error) {
	logger = logger.Session("db-cancel-task", lager.Data{"task_guid": taskGuid})
	logger.Info("starting")
	defer logger.Info("complete")
	var beforeTask models.Task
	var afterTask *models.Task
	var cellID string
	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		var err error
		afterTask, err = db.fetchTaskForUpdate(ctx, logger, taskGuid, tx)
		if err != nil {
			logger.Error("failed-locking-task", err)
			return err
		}
		beforeTask = *afterTask
		cellID = afterTask.CellId
		if err = afterTask.ValidateTransitionTo(models.Task_Completed); err != nil {
			if afterTask.State != models.Task_Pending {
				logger.Error("failed-to-transition-task-to-completed", err)
				return err
			}
		}
		err = db.completeTask(ctx, logger, afterTask, true, "task was cancelled", "", tx)
		if err != nil {
			return err
		}
		return nil
	})
	return &beforeTask, afterTask, cellID, err
}
func (db *SQLDB) CompleteTask(ctx context.Context, logger lager.Logger, taskGuid, cellID string, failed bool, failureReason, taskResult string) (*models.Task, *models.Task, error) {
	logger = logger.Session("db-complete-task", lager.Data{"task_guid": taskGuid, "cell_id": cellID})
	logger.Info("starting")
	defer logger.Info("complete")
	var beforeTask models.Task
	var afterTask *models.Task
	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		var err error
		afterTask, err = db.fetchTaskForUpdate(ctx, logger, taskGuid, tx)
		if err != nil {
			logger.Error("failed-locking-task", err)
			return err
		}
		beforeTask = *afterTask
		if afterTask.CellId != cellID && afterTask.State == models.Task_Running {
			logger.Error("failed-task-already-running-on-different-cell", err)
			return models.NewRunningOnDifferentCellError(cellID, afterTask.CellId)
		}
		if err = afterTask.ValidateTransitionTo(models.Task_Completed); err != nil {
			logger.Error("failed-to-transition-task-to-completed", err)
			return err
		}
		err = db.completeTask(ctx, logger, afterTask, failed, failureReason, taskResult, tx)
		if err != nil {
			return err
		}
		return nil
	})
	return &beforeTask, afterTask, err
}
func (db *SQLDB) FailTask(ctx context.Context, logger lager.Logger, taskGuid, failureReason string) (*models.Task, *models.Task, error) {
	logger = logger.Session("db-fail-task", lager.Data{"task_guid": taskGuid})
	logger.Info("starting")
	defer logger.Info("complete")
	var beforeTask models.Task
	var afterTask *models.Task
	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		var err error
		afterTask, err = db.fetchTaskForUpdate(ctx, logger, taskGuid, tx)
		if err != nil {
			logger.Error("failed-locking-task", err)
			return err
		}
		beforeTask = *afterTask
		if err = afterTask.ValidateTransitionTo(models.Task_Completed); err != nil {
			if afterTask.State != models.Task_Pending {
				logger.Error("failed-to-transition-task-to-completed", err)
				return err
			}
		}
		err = db.completeTask(ctx, logger, afterTask, true, failureReason, "", tx)
		if err != nil {
			return err
		}
		return nil
	})
	return &beforeTask, afterTask, err
}
func (db *SQLDB) RejectTask(ctx context.Context, logger lager.Logger, taskGuid, rejectionReason string) (*models.Task, *models.Task, error) {
	logger = logger.Session("db-reject-task", lager.Data{"task_guid": taskGuid})
	logger.Info("starting")
	defer logger.Info("complete")
	var beforeTask models.Task
	var afterTask *models.Task
	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
		var err error
		afterTask, err = db.fetchTaskForUpdate(ctx, logger, taskGuid, tx)
		if err != nil {
			logger.Error("failed-locking-task", err)
			return err
		}
		if afterTask.State != models.Task_Pending && afterTask.State != models.Task_Running {
			logger.Info("invalid-task-state", lager.Data{"task_state": afterTask.State})
			return models.ErrBadRequest
		}
		beforeTask = *afterTask
		now := db.clock.Now().UnixNano()
		afterTask.RejectionCount++
		afterTask.RejectionReason = truncateString(rejectionReason, 1024)
		afterTask.State = models.Task_Pending
		afterTask.UpdatedAt = now
		_, err = db.update(ctx, logger, tx, tasksTable,
			helpers.SQLAttributes{
				"rejection_count":  afterTask.RejectionCount,
				"rejection_reason": afterTask.RejectionReason,
				"updated_at":       afterTask.UpdatedAt,
				"state":            afterTask.State,
			},
			"guid = ?", taskGuid,
		)
		if err != nil {
			logger.Error("failed-updating-tasks", err)
			return err
		}
		return nil
	})
	return &beforeTask, afterTask, err
}