diff --git a/api/job.go b/api/job.go index 8223b04..8210da3 100644 --- a/api/job.go +++ b/api/job.go @@ -5,6 +5,7 @@ import ( "errors" "net/http" "strconv" + "strings" "time" "github.com/gaucho-racing/foreman/model" @@ -187,11 +188,33 @@ func FailRun(c *gin.Context) { // ---------- Job reads ---------- -// jobWithRun is the response shape when ?include=current_run is set on -// /jobs or /jobs/:id. CurrentRun is null when no in-flight run exists. +// jobWithRun is the response shape when ?include= is set on /jobs or +// /jobs/:id. CurrentRun is null when no in-flight run exists. LastRun is +// omitted entirely unless asked for, so include=current_run responses stay +// byte-identical to what they were before last_run existed. type jobWithRun struct { model.Job CurrentRun *model.JobRun `json:"current_run"` + LastRun *model.JobRun `json:"last_run,omitempty"` +} + +// includeSet parses the comma-separated ?include= param. Two values are +// recognized: +// +// - current_run — the in-flight attempt; null unless the job is active. +// - last_run — the newest attempt whatever its status, so pending and +// terminal jobs still carry progress / error / result. +// +// Unknown values are ignored rather than rejected, so adding one later +// can't break a client that already sends it. +func includeSet(c *gin.Context) map[string]bool { + out := map[string]bool{} + for _, part := range strings.Split(c.Query("include"), ",") { + if p := strings.TrimSpace(part); p != "" { + out[p] = true + } + } + return out } func GetJob(c *gin.Context) { @@ -199,16 +222,29 @@ func GetJob(c *gin.Context) { if respondServiceErr(c, err) { return } - if c.Query("include") != "current_run" { + inc := includeSet(c) + if !inc["current_run"] && !inc["last_run"] { c.JSON(http.StatusOK, job) return } - run, err := service.CurrentRun(job.ID) - if err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) - return + out := jobWithRun{Job: job} + if inc["current_run"] { + run, err := service.CurrentRun(job.ID) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + out.CurrentRun = run + } + if inc["last_run"] { + run, err := service.LastRun(job.ID) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + out.LastRun = run } - c.JSON(http.StatusOK, jobWithRun{Job: job, CurrentRun: run}) + c.JSON(http.StatusOK, out) } func ListJobs(c *gin.Context) { @@ -225,7 +261,8 @@ func ListJobs(c *gin.Context) { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } - if c.Query("include") != "current_run" { + inc := includeSet(c) + if !inc["current_run"] && !inc["last_run"] { c.JSON(http.StatusOK, jobs) return } @@ -233,23 +270,43 @@ func ListJobs(c *gin.Context) { for i, j := range jobs { ids[i] = j.ID } - runs, err := service.CurrentRunsForJobs(ids) - if err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) - return + var current, last map[string]model.JobRun + if inc["current_run"] { + current, err = service.CurrentRunsForJobs(ids) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + } + if inc["last_run"] { + last, err = service.LastRunsForJobs(ids) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } } out := make([]jobWithRun, len(jobs)) for i, j := range jobs { - var run *model.JobRun - if r, ok := runs[j.ID]; ok { - r := r // local copy so the pointer is stable across loop iterations - run = &r + out[i] = jobWithRun{ + Job: j, + CurrentRun: runFor(current, j.ID), + LastRun: runFor(last, j.ID), } - out[i] = jobWithRun{Job: j, CurrentRun: run} } c.JSON(http.StatusOK, out) } +// runFor pulls a job's run out of a batch map as a pointer. The local copy +// matters: taking &m[id] isn't allowed for maps, and reusing a loop variable's +// address would leave every row pointing at the last one. +func runFor(m map[string]model.JobRun, jobID string) *model.JobRun { + r, ok := m[jobID] + if !ok { + return nil + } + return &r +} + // ListJobRuns returns every attempt at a job, oldest first. 404s match // GetJob: a missing job id returns 404 (instead of an empty list) so // callers can disambiguate "no runs yet" from "wrong id". diff --git a/api/sse.go b/api/sse.go index 100905b..15da2ae 100644 --- a/api/sse.go +++ b/api/sse.go @@ -57,12 +57,20 @@ func StreamJobEvents(c *gin.Context) { } } -// buildJobEvent wraps a Job with its in-flight Run for SSE consumers. -// Reuses the jobWithRun shape that /jobs?include=current_run already -// returns so the dashboard's decode path is identical for both. Errors -// looking up the run are swallowed — better to push the bare job than -// drop the event entirely. +// buildJobEvent wraps a Job with its Run for SSE consumers. Reuses the +// jobWithRun shape that /jobs?include= already returns so the dashboard's +// decode path is identical for both. Errors looking up the run are +// swallowed — better to push the bare job than drop the event entirely. +// +// One query serves both fields: the newest attempt IS the in-flight one +// whenever a job is running, so current_run keeps its exact old meaning +// (non-null only while an attempt holds the lease) while last_run also +// carries the final reading once the job stops. func buildJobEvent(job model.Job) jobWithRun { - run, _ := service.CurrentRun(job.ID) - return jobWithRun{Job: job, CurrentRun: run} + last, _ := service.LastRun(job.ID) + ev := jobWithRun{Job: job, LastRun: last} + if last != nil && last.Status == model.RunStatusRunning { + ev.CurrentRun = last + } + return ev } diff --git a/clients/go/foreman.go b/clients/go/foreman.go index 6d399fe..0b41d64 100644 --- a/clients/go/foreman.go +++ b/clients/go/foreman.go @@ -78,6 +78,11 @@ type Job struct { // CurrentRun is populated only when a Get/List was called with // include=current_run. nil otherwise. CurrentRun *Run `json:"current_run,omitempty"` + // LastRun is populated only when a Get/List was called with + // include=last_run. Unlike CurrentRun it survives the attempt + // finishing, so it still carries progress / error / result on a + // pending-after-retry or terminal job. nil otherwise. + LastRun *Run `json:"last_run,omitempty"` } // Run is one attempt at a Job. Each Claim creates a new Run. @@ -190,6 +195,7 @@ type JobsFilter struct { Limit int Cursor string IncludeCurrentRun bool + IncludeLastRun bool } type RunsFilter struct { @@ -429,8 +435,17 @@ func (c *Client) ListJobs(ctx context.Context, f JobsFilter) ([]Job, error) { if f.Cursor != "" { q.Set("cursor", f.Cursor) } + // The server parses ?include= as a comma-separated set, so asking for + // both in one call is a single request. + var include []string if f.IncludeCurrentRun { - q.Set("include", "current_run") + include = append(include, "current_run") + } + if f.IncludeLastRun { + include = append(include, "last_run") + } + if len(include) > 0 { + q.Set("include", strings.Join(include, ",")) } var out []Job if err := c.simpleJSON(ctx, http.MethodGet, "/foreman/jobs", queryParams{q: q}, &out, "list-jobs"); err != nil { diff --git a/service/job.go b/service/job.go index 899b32b..0c02834 100644 --- a/service/job.go +++ b/service/job.go @@ -512,6 +512,26 @@ func CurrentRun(jobID string) (*model.JobRun, error) { return &run, nil } +// LastRun returns the most recent attempt at a job whatever its status, +// so callers can read a finished attempt's progress / error / result. +// CurrentRun only ever matches a running attempt, which leaves pending +// (bounced, awaiting re-claim) and terminal jobs with nothing to show. +// Returns nil when the job has never been claimed. +func LastRun(jobID string) (*model.JobRun, error) { + var run model.JobRun + err := database.DB. + Where("job_id = ?", jobID). + Order("attempt DESC"). + First(&run).Error + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + return nil, err + } + return &run, nil +} + // ListRuns returns every attempt at the given job, oldest first. func ListRuns(jobID string) ([]model.JobRun, error) { var runs []model.JobRun @@ -672,3 +692,27 @@ func CurrentRunsForJobs(jobIDs []string) (map[string]model.JobRun, error) { } return out, nil } + +// LastRunsForJobs batches LastRun across many jobs in one query, for the +// jobs-list ?include=last_run expansion. DISTINCT ON keeps a single row +// per job_id — the highest attempt — so a 50-row list costs one query +// rather than one per job. Returns a map keyed by job_id; jobs that have +// never been claimed simply have no entry. +func LastRunsForJobs(jobIDs []string) (map[string]model.JobRun, error) { + out := make(map[string]model.JobRun, len(jobIDs)) + if len(jobIDs) == 0 { + return out, nil + } + var runs []model.JobRun + sql := fmt.Sprintf( + `SELECT DISTINCT ON (job_id) * FROM %s WHERE job_id IN ? ORDER BY job_id, attempt DESC`, + model.TableJobRuns(), + ) + if err := database.DB.Raw(sql, jobIDs).Scan(&runs).Error; err != nil { + return nil, err + } + for _, r := range runs { + out[r.JobID] = r + } + return out, nil +} diff --git a/service/job_test.go b/service/job_test.go index 5291885..379ccc8 100644 --- a/service/job_test.go +++ b/service/job_test.go @@ -361,6 +361,129 @@ func TestFail_OnCancelRequestedTerminalizesAsCancelled(t *testing.T) { } } +// ---------- LastRun / LastRunsForJobs ---------- + +func TestLastRun_NilWhenNeverClaimed(t *testing.T) { + resetDB(t) + job := mustEnqueue(t, "k") + run, err := LastRun(job.ID) + if err != nil { + t.Fatal(err) + } + if run != nil { + t.Fatalf("expected nil for an unclaimed job, got attempt %d", run.Attempt) + } +} + +// The whole point of LastRun: CurrentRun goes nil the moment an attempt +// finishes, so a bounced job has nothing to show without this. +func TestLastRun_SurvivesFailedAttemptWhereCurrentRunDoesNot(t *testing.T) { + resetDB(t) + _ = mustEnqueue(t, "k", withMaxAttempts(3)) + res, _, _ := Claim(ClaimParams{Kinds: []string{"k"}, WorkerID: "w-1", LeaseSec: 30}) + if _, err := Fail(res.Run.ID, "w-1", "boom", true, 0, []byte(`{"rows":7}`)); err != nil { + t.Fatal(err) + } + + current, err := CurrentRun(res.Job.ID) + if err != nil { + t.Fatal(err) + } + if current != nil { + t.Fatalf("CurrentRun should be nil after the attempt failed, got %s", current.Status) + } + + last, err := LastRun(res.Job.ID) + if err != nil { + t.Fatal(err) + } + if last == nil { + t.Fatal("LastRun must return the failed attempt") + } + if last.Status != model.RunStatusFailed || last.Error != "boom" { + t.Fatalf("expected the failed attempt, got status=%s error=%q", last.Status, last.Error) + } + jsonEqual(t, last.Result, []byte(`{"rows":7}`)) +} + +func TestLastRun_ReturnsHighestAttempt(t *testing.T) { + resetDB(t) + _ = mustEnqueue(t, "k", withMaxAttempts(3)) + first, _, _ := Claim(ClaimParams{Kinds: []string{"k"}, WorkerID: "w-1", LeaseSec: 30}) + if _, err := Fail(first.Run.ID, "w-1", "first", true, 0, nil); err != nil { + t.Fatal(err) + } + second, found, err := Claim(ClaimParams{Kinds: []string{"k"}, WorkerID: "w-2", LeaseSec: 30}) + if err != nil || !found { + t.Fatalf("re-claim: found=%v err=%v", found, err) + } + + last, err := LastRun(second.Job.ID) + if err != nil { + t.Fatal(err) + } + if last == nil || last.Attempt != 2 { + t.Fatalf("expected attempt 2, got %+v", last) + } + // Attempt 2 is in flight, so here last_run and current_run agree. + if last.Status != model.RunStatusRunning { + t.Fatalf("expected the running attempt, got %s", last.Status) + } +} + +func TestLastRunsForJobs_OneRowPerJobAtHighestAttempt(t *testing.T) { + resetDB(t) + // Job A: two attempts, newest failed. Job B: one running attempt. + // Job C: never claimed, so it must be absent from the map. + _ = mustEnqueue(t, "a", withMaxAttempts(3)) + a1, _, _ := Claim(ClaimParams{Kinds: []string{"a"}, WorkerID: "w-1", LeaseSec: 30}) + if _, err := Fail(a1.Run.ID, "w-1", "a-first", true, 0, nil); err != nil { + t.Fatal(err) + } + a2, _, _ := Claim(ClaimParams{Kinds: []string{"a"}, WorkerID: "w-1", LeaseSec: 30}) + if _, err := Fail(a2.Run.ID, "w-1", "a-second", true, 0, nil); err != nil { + t.Fatal(err) + } + b := mustClaimNew(t, "b", "w-2") + c := mustEnqueue(t, "c") + + got, err := LastRunsForJobs([]string{a2.Job.ID, b.Job.ID, c.ID}) + if err != nil { + t.Fatal(err) + } + if len(got) != 2 { + t.Fatalf("expected 2 entries (c was never claimed), got %d: %+v", len(got), got) + } + ra, ok := got[a2.Job.ID] + if !ok { + t.Fatal("job a missing") + } + if ra.Attempt != 2 || ra.Error != "a-second" { + t.Fatalf("job a: expected attempt 2 / a-second, got attempt %d / %q", ra.Attempt, ra.Error) + } + rb, ok := got[b.Job.ID] + if !ok { + t.Fatal("job b missing") + } + if rb.Status != model.RunStatusRunning { + t.Fatalf("job b: expected running, got %s", rb.Status) + } + if _, ok := got[c.ID]; ok { + t.Fatal("job c has no runs and must not appear in the map") + } +} + +func TestLastRunsForJobs_EmptyInputNoQuery(t *testing.T) { + resetDB(t) + got, err := LastRunsForJobs(nil) + if err != nil { + t.Fatal(err) + } + if len(got) != 0 { + t.Fatalf("expected empty map, got %+v", got) + } +} + // ---------- helpers ---------- func mustClaimNew(t *testing.T, kind, worker string) ClaimResult {