Bug: Fix pipeline run complete state machine

.changelog/4053.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+pipelines: Properly mark a pipeline run as complete
+```

internal/server/boltdbstate/job.go

@@ -1772,11 +1772,25 @@ func (s *State) pipelineComplete(jobId string) error {
  if job.State == pb.Job_ERROR {
  run.State = pb.PipelineRun_ERROR
  } else if job.State == pb.Job_SUCCESS {
- // If job Id matches last job queued by pipeline.
- // We will have to change this in the future when pipeline steps run in parallel,
- // and the last job queued may not be the last job to complete in the pipeline
- // TODO:XX figure out how ^
- if job.Id == run.Jobs[len(run.Jobs)-1].Id {
+ // Look at all job ids in a run and check if any are not SUCCESS
+ runComplete := true
+ for _, j := range run.Jobs {
+ if j.Id == job.Id {
+ continue
+ }
+
+ rj, err := s.JobById(j.Id, nil)
+ if err != nil {
+ return err
+ }
+
+ if rj.State != pb.Job_SUCCESS {
+ runComplete = false
+ break
+ }
+ }
+
+ if runComplete {
  run.State = pb.PipelineRun_SUCCESS
  s.log.Trace("pipeline run is complete", "job", job.Id, "pipeline", job.Pipeline.PipelineId, "run", run.Sequence)
  }

pkg/server/gen/server.swagger.json

@@ -9477,16 +9477,19 @@
  "description": "The sequence number for this pipeline run."
  },
  "pipeline": {
- "$ref": "#/definitions/hashicorp.waypoint.Ref.Pipeline"
+ "$ref": "#/definitions/hashicorp.waypoint.Ref.Pipeline",
+ "description": "The pipeline associated with this run."
  },
  "jobs": {
  "type": "array",
  "items": {
  "$ref": "#/definitions/hashicorp.waypoint.Ref.Job"
- }
+ },
+ "description": "The full list of jobs that are associated with this run."
  },
  "state": {
- "$ref": "#/definitions/hashicorp.waypoint.PipelineRun.State"
+ "$ref": "#/definitions/hashicorp.waypoint.PipelineRun.State",
+ "title": "The current state of this pipeline run,"
  }
  }
  },

pkg/server/proto/server.proto

@@ -4870,10 +4870,13 @@ message PipelineRun {
  // The sequence number for this pipeline run.
  uint64 sequence = 2;
 
+ // The pipeline associated with this run.
  Ref.Pipeline pipeline = 3;
 
+ // The full list of jobs that are associated with this run.
  repeated Ref.Job jobs = 4;
 
+ // The current state of this pipeline run,
  State state = 5;
 
  enum State {

pkg/server/singleprocess/service_pipeline.go

@@ -512,6 +512,7 @@ func (s *Service) buildStepJobs(
  }
  }
 
+ // Include a list of all associated jobs for this specific run
  pipelineRun.Jobs = append(pipelineRun.Jobs, &pb.Ref_Job{Id: job.Id})
  stepJobs = append(stepJobs, &pb.QueueJobRequest{
  Job: job,