ExecutionTimeout to fail executions instead of jobs

pkg/jobstore/boltdb/store_test.go

@@ -107,6 +107,9 @@ func (s *BoltJobstoreTestSuite) SetupTest() {
 		s.clock.Add(1 * time.Second)
 		execution := mock.ExecutionForJob(job)
 		execution.ComputeState.StateType = models.ExecutionStateNew
+		// clear out CreateTime and ModifyTime from the mocked execution to let the job store fill those
+		execution.CreateTime = 0
+		execution.ModifyTime = 0
 		err = s.store.CreateExecution(s.ctx, *execution, models.Event{})
 		s.Require().NoError(err)
 

pkg/models/evaluation.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/bacalhau-project/bacalhau/pkg/lib/math"
+	"github.com/bacalhau-project/bacalhau/pkg/util/idgen"
 )
 
 const (
@@ -16,10 +17,11 @@ const (
 )
 
 const (
-	EvalTriggerJobRegister     = "job-register"
-	EvalTriggerJobCancel       = "job-cancel"
-	EvalTriggerRetryFailedExec = "exec-failure"
-	EvalTriggerExecUpdate      = "exec-update"
+	EvalTriggerJobRegister = "job-register"
+	EvalTriggerJobCancel   = "job-cancel"
+	EvalTriggerExecFailure = "exec-failure"
+	EvalTriggerExecUpdate  = "exec-update"
+	EvalTriggerExecTimeout = "exec-timeout"
 )
 
 // Evaluation is just to ask the scheduler to reassess if additional job instances must be
@@ -61,6 +63,81 @@ type Evaluation struct {
 	ModifyTime int64 `json:"ModifyTime"`
 }
 
+// NewEvaluation creates a new Evaluation.
+func NewEvaluation() *Evaluation {
+	return &Evaluation{
+		ID:         idgen.NewEvaluationID(),
+		Status:     EvalStatusPending,
+		CreateTime: time.Now().UTC().UnixNano(),
+		ModifyTime: time.Now().UTC().UnixNano(),
+	}
+}
+
+// WithJobID sets the JobID of the Evaluation.
+func (e *Evaluation) WithJobID(jobID string) *Evaluation {
+	e.JobID = jobID
+	return e
+}
+
+// WithNamespace sets the Namespace of the Evaluation.
+func (e *Evaluation) WithNamespace(namespace string) *Evaluation {
+	e.Namespace = namespace
+	return e
+}
+
+// WithTriggeredBy sets the TriggeredBy of the Evaluation.
+func (e *Evaluation) WithTriggeredBy(triggeredBy string) *Evaluation {
+	e.TriggeredBy = triggeredBy
+	return e
+}
+
+// WithPriority sets the Priority of the Evaluation.
+func (e *Evaluation) WithPriority(priority int) *Evaluation {
+	e.Priority = priority
+	return e
+}
+
+// WithType sets the Type of the Evaluation.
+func (e *Evaluation) WithType(jobType string) *Evaluation {
+	e.Type = jobType
+	return e
+}
+
+// WithStatus sets the Status of the Evaluation.
+func (e *Evaluation) WithStatus(status string) *Evaluation {
+	e.Status = status
+	return e
+}
+
+// WithComment sets the Comment of the Evaluation.
+func (e *Evaluation) WithComment(comment string) *Evaluation {
+	e.Comment = comment
+	return e
+}
+
+// WithWaitUntil sets the WaitUntil of the Evaluation.
+func (e *Evaluation) WithWaitUntil(waitUntil time.Time) *Evaluation {
+	e.WaitUntil = waitUntil
+	return e
+}
+
+// Normalize ensures that the Evaluation is in a valid state.
+func (e *Evaluation) Normalize() *Evaluation {
+	if e.ID == "" {
+		e.ID = idgen.NewEvaluationID()
+	}
+	if e.Status == "" {
+		e.Status = EvalStatusPending
+	}
+	if e.CreateTime == 0 {
+		e.CreateTime = time.Now().UTC().UnixNano()
+	}
+	if e.ModifyTime == 0 {
+		e.ModifyTime = time.Now().UTC().UnixNano()
+	}
+	return e
+}
+
 // TerminalStatus returns if the current status is terminal and
 // will no longer transition.
 func (e *Evaluation) TerminalStatus() bool {

pkg/models/event.go

@@ -57,7 +57,12 @@ func (e *Event) WithMessage(message string) *Event {
 
 // WithError returns a new Event with the given error.
 func (e *Event) WithError(err error) *Event {
-	e.Message = err.Error()
+	return e.WithErrorMessage(err.Error())
+}
+
+// WithErrorMessage returns a new Event with the given error.
+func (e *Event) WithErrorMessage(message string) *Event {
+	e.Message = message
 	return e.WithDetail(DetailsKeyIsError, "true")
 }
 

pkg/models/execution.go

@@ -135,6 +135,12 @@ func (e *Execution) GetModifyTime() time.Time {
 	return time.Unix(0, e.ModifyTime).UTC()
 }
 
+// HasExecutionExpired returns true if the execution has been running longer than
+// the provided timeout duration.
+func (e *Execution) HasExecutionExpired(timeout time.Duration) bool {
+	return e.ComputeState.StateType == ExecutionStateBidAccepted && time.Since(e.GetModifyTime()) > timeout
+}
+
 // Normalize Allocation to ensure fields are initialized to the expectations
 // of this version of Bacalhau. Should be called when restoring persisted
 // Executions or receiving Executions from Bacalhau clients potentially on an

pkg/node/config_defaults.go

@@ -47,6 +47,7 @@ var DefaultRequesterConfig = RequesterConfigParams{
 	},
 
 	HousekeepingBackgroundTaskInterval: 30 * time.Second,
+	HousekeepingTimeoutBuffer:          2 * time.Minute,
 	NodeRankRandomnessRange:            5,
 	OverAskForBidsFactor:               3,
 
@@ -83,6 +84,7 @@ var TestRequesterConfig = RequesterConfigParams{
 		ExecutionTimeout: 30 * time.Second,
 	},
 	HousekeepingBackgroundTaskInterval: 30 * time.Second,
+	HousekeepingTimeoutBuffer:          100 * time.Millisecond,
 	NodeRankRandomnessRange:            5,
 	OverAskForBidsFactor:               3,
 

pkg/node/config_requester.go

@@ -21,6 +21,7 @@ type RequesterConfigParams struct {
 	JobDefaults transformer.JobDefaults
 
 	HousekeepingBackgroundTaskInterval time.Duration
+	HousekeepingTimeoutBuffer          time.Duration
 	NodeRankRandomnessRange            int
 	OverAskForBidsFactor               uint
 	JobSelectionPolicy                 JobSelectionPolicy

pkg/node/requester.go

@@ -249,12 +249,16 @@ func NewRequesterNode(
 		ResultTransformer: resultTransformers,
 	})
 
-	housekeeping := requester.NewHousekeeping(requester.HousekeepingParams{
-		Endpoint: endpoint,
-		JobStore: jobStore,
-		NodeID:   nodeID,
-		Interval: requesterConfig.HousekeepingBackgroundTaskInterval,
+	housekeeping, err := orchestrator.NewHousekeeping(orchestrator.HousekeepingParams{
+		EvaluationBroker: evalBroker,
+		JobStore:         jobStore,
+		Interval:         requesterConfig.HousekeepingBackgroundTaskInterval,
+		TimeoutBuffer:    requesterConfig.HousekeepingTimeoutBuffer,
 	})
+	if err != nil {
+		return nil, err
+	}
+	housekeeping.Start(ctx)
 
 	// register debug info providers for the /debug endpoint
 	debugInfoProviders := []model.DebugInfoProvider{
@@ -301,7 +305,7 @@ func NewRequesterNode(
 	// A single Cleanup function to make sure the order of closing dependencies is correct
 	cleanupFunc := func(ctx context.Context) {
 		// stop the housekeeping background task
-		housekeeping.Stop()
+		housekeeping.Stop(ctx)
 		for _, worker := range workers {
 			worker.Stop()
 		}

pkg/orchestrator/events.go

@@ -1,14 +1,16 @@
 package orchestrator
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 )
 
 const (
-	EventTopicJobSubmission models.EventTopic = "Submission"
-	EventTopicJobScheduling models.EventTopic = "Scheduling"
+	EventTopicJobSubmission    models.EventTopic = "Submission"
+	EventTopicJobScheduling    models.EventTopic = "Scheduling"
+	EventTopicExecutionTimeout models.EventTopic = "Exec Timeout"
 )
 
 const (
@@ -23,6 +25,9 @@ const (
 	execStoppedByOversubscriptionMessage = "Execution stop requested because there are more executions than needed"
 	execRejectedByNodeMessage            = "Node responded to execution run request"
 	execFailedMessage                    = "Execution did not complete successfully"
+
+	executionTimeoutMessage = "Execution timed out"
+	executionTimeoutHint    = "Try increasing the task timeout or reducing the task size"
 )
 
 func event(topic models.EventTopic, msg string, details map[string]string) models.Event {
@@ -63,6 +68,14 @@ func ExecStoppedByNodeUnhealthyEvent() models.Event {
 	return event(EventTopicJobScheduling, execStoppedByNodeUnhealthyMessage, map[string]string{})
 }
 
+func ExecStoppedByExecutionTimeoutEvent(timeout time.Duration) models.Event {
+	e := models.NewEvent(EventTopicExecutionTimeout).
+		WithErrorMessage(fmt.Sprintf("%s. Execution took longer than %s", executionTimeoutMessage, timeout)).
+		WithHint(executionTimeoutHint).
+		WithFailsExecution(true)
+	return *e
+}
+
 func ExecStoppedByNodeRejectedEvent() models.Event {
 	return event(EventTopicJobScheduling, execStoppedByNodeRejectedMessage, map[string]string{})
 }