fix: RP error handling (#514)

pkg/executor/bacalhau/bacalhau.go

@@ -128,6 +128,13 @@ func (executor *BacalhauExecutor) RunJob(
 			if jobInfo.Job.State.StateType == models.JobStateTypeCompleted {
 				break
 			}
+
+			// Jobs have 3 terminal states: completed, stopped and failed. We catch the latter two here.
+			// Most likely, the job failed, but could have been stopped by the operator.
+			// Any other state, the job is still running or yet to be run.
+			if jobInfo.Job.State.StateType.IsTerminal() {
+				return nil, fmt.Errorf("job failed to execute %s : %s", jobID, jobInfo.Job.State.Message)
+			}
 		}
 
 		time.Sleep(time.Duration(executor.Options.JobStatusPollInterval) * time.Second)

pkg/resourceprovider/controller.go

@@ -436,6 +436,11 @@ func (controller *ResourceProviderController) runJobs(ctx context.Context) error
 // we've already updated controller.runningJobs so we know this will only
 // run once
 func (controller *ResourceProviderController) runJob(ctx context.Context, deal data.DealContainer) {
+	defer func() {
+		// Everytime we finish a job we need to ensure we have resource offers
+		controller.ensureResourceOffers()
+	}()
+
 	controller.log.Info("run job", deal)
 	controller.log.Info("deal ID", deal.Deal.ID)
 
@@ -551,7 +556,5 @@ func (controller *ResourceProviderController) runJob(ctx context.Context, deal d
 	}
 	span.AddEvent("solver.transaction_hash.added")
 
-	controller.ensureResourceOffers()
-
 	span.AddEvent("done")
 }