From 8c291a2934568cb3dfa96d2de0eaae4a634e96a1 Mon Sep 17 00:00:00 2001 From: Michael Gaffney Date: Tue, 24 Sep 2024 09:27:24 -0400 Subject: [PATCH] feat(scheduler): Add retries to job status This adds retries as an attribute to JobStatus allowing a job to indicate liveness to the scheduler when work has not progressed. A job could use this and the statusThreshold passed into the Run method to break work up into batches that would otherwise timeout if all the work was attempted at once, for example a long running SQL delete command. --- .../92/04_add_column_to_job_run.up.sql | 11 ++++++ .../controller/storage/job/store/v1/job.proto | 5 +++ .../scheduler/additional_verification_test.go | 3 +- internal/scheduler/job.go | 6 ++++ .../job/additional_verification_test.go | 6 ++-- internal/scheduler/job/query.go | 5 ++- internal/scheduler/job/repository_run.go | 12 +++---- internal/scheduler/job/repository_run_test.go | 34 ++++++++++-------- internal/scheduler/job/store/job.pb.go | 35 +++++++++++++------ internal/scheduler/scheduler.go | 8 ++--- 10 files changed, 85 insertions(+), 40 deletions(-) create mode 100644 internal/db/schema/migrations/oss/postgres/92/04_add_column_to_job_run.up.sql diff --git a/internal/db/schema/migrations/oss/postgres/92/04_add_column_to_job_run.up.sql b/internal/db/schema/migrations/oss/postgres/92/04_add_column_to_job_run.up.sql new file mode 100644 index 000000000..9a077cc02 --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/92/04_add_column_to_job_run.up.sql @@ -0,0 +1,11 @@ +-- Copyright (c) HashiCorp, Inc. +-- SPDX-License-Identifier: BUSL-1.1 + +begin; + + alter table job_run + add column retries_count int not null default 0 + constraint retries_count_can_not_be_negative + check(retries_count >= 0); + +commit; diff --git a/internal/proto/controller/storage/job/store/v1/job.proto b/internal/proto/controller/storage/job/store/v1/job.proto index f3380740d..949c24927 100644 --- a/internal/proto/controller/storage/job/store/v1/job.proto +++ b/internal/proto/controller/storage/job/store/v1/job.proto @@ -67,6 +67,11 @@ message JobRun { // @inject_tag: `gorm:"default:0"` uint32 total_count = 9; + // retries_count is set during an update to indicate the number of times + // a job has retried work. + // @inject_tag: `gorm:"default:0"` + uint32 retries_count = 12; + // status of the job run (running, completed, failed or interrupted). // @inject_tag: `gorm:"not_null"` string status = 10; diff --git a/internal/scheduler/additional_verification_test.go b/internal/scheduler/additional_verification_test.go index 6d872bed0..cacab68ea 100644 --- a/internal/scheduler/additional_verification_test.go +++ b/internal/scheduler/additional_verification_test.go @@ -317,7 +317,7 @@ func TestSchedulerJobProgress(t *testing.T) { <-statusRequest // Send progress to monitor loop to persist - jobStatus <- JobStatus{Total: 10, Completed: 0} + jobStatus <- JobStatus{Total: 10, Completed: 0, Retries: 1} // Wait for scheduler to query for job status before verifying previous results <-statusRequest @@ -329,6 +329,7 @@ func TestSchedulerJobProgress(t *testing.T) { assert.Equal(string(job.Running), run.Status) assert.Equal(uint32(10), run.TotalCount) assert.Equal(uint32(0), run.CompletedCount) + assert.Equal(uint32(1), run.RetriesCount) // Send progress to monitor loop to persist jobStatus <- JobStatus{Total: 20, Completed: 10} diff --git a/internal/scheduler/job.go b/internal/scheduler/job.go index bbdcf7e10..bbb2baff1 100644 --- a/internal/scheduler/job.go +++ b/internal/scheduler/job.go @@ -60,6 +60,9 @@ type Job interface { // The Completed value cannot be greater than the Total value // and both values must be greater than or equal to zero. // +// Retries is used to indicate how many times the job has retried +// accomplishing work. The value must be greater than or equal to zero. +// // The scheduler uses the values in the JobStatus to verify that // progress is being made by the job. The scheduler will interrupt the job // if the values returned by Status do not change over a configurable @@ -68,6 +71,9 @@ type JobStatus struct { // The job's work items Completed int // number of items processed Total int // total number of items to be processed + + // The job's liveliness + Retries int // number of times the job has retried work } func validateJob(ctx context.Context, j Job) error { diff --git a/internal/scheduler/job/additional_verification_test.go b/internal/scheduler/job/additional_verification_test.go index 4e221b14f..0e9c351fd 100644 --- a/internal/scheduler/job/additional_verification_test.go +++ b/internal/scheduler/job/additional_verification_test.go @@ -44,7 +44,7 @@ func TestJobWorkflow(t *testing.T) { assert.NotEmpty(run.PrivateId) assert.Equal(job.Name, run.JobName) - run, err = repo.UpdateProgress(ctx, run.PrivateId, 100, 110) + run, err = repo.UpdateProgress(ctx, run.PrivateId, 100, 110, 0) require.NoError(err) assert.Equal(uint32(100), run.CompletedCount) assert.Equal(uint32(110), run.TotalCount) @@ -54,7 +54,7 @@ func TestJobWorkflow(t *testing.T) { require.NoError(err) assert.Nil(newRuns) - run, err = repo.CompleteRun(ctx, run.PrivateId, time.Hour, 0, 0) + run, err = repo.CompleteRun(ctx, run.PrivateId, time.Hour, 0, 0, 0) require.NoError(err) assert.Equal(Completed.string(), run.Status) @@ -85,7 +85,7 @@ func TestJobWorkflow(t *testing.T) { require.NoError(err) assert.Nil(newRuns) - newRun, err = repo.FailRun(ctx, newRun.PrivateId, 0, 0) + newRun, err = repo.FailRun(ctx, newRun.PrivateId, 0, 0, 0) require.NoError(err) assert.Equal(Failed.string(), newRun.Status) diff --git a/internal/scheduler/job/query.go b/internal/scheduler/job/query.go index dbb8be5fb..41bd0c318 100644 --- a/internal/scheduler/job/query.go +++ b/internal/scheduler/job/query.go @@ -65,7 +65,8 @@ const updateProgressQuery = ` job_run set completed_count = ?, - total_count = ? + total_count = ?, + retries_count = ? where private_id = ? and status = 'running' @@ -78,6 +79,7 @@ const completeRunQuery = ` set completed_count = ?, total_count = ?, + retries_count = ?, status = 'completed', end_time = current_timestamp where @@ -92,6 +94,7 @@ const failRunQuery = ` set completed_count = ?, total_count = ?, + retries_count = ?, status = 'failed', end_time = current_timestamp where diff --git a/internal/scheduler/job/repository_run.go b/internal/scheduler/job/repository_run.go index d0f54e712..e31044e99 100644 --- a/internal/scheduler/job/repository_run.go +++ b/internal/scheduler/job/repository_run.go @@ -71,7 +71,7 @@ func (r *Repository) RunJobs(ctx context.Context, serverId string, opt ...Option // Once a run has been persisted with a final run status (completed, failed or interrupted), // any future UpdateProgress attempts will return an error with Code errors.InvalidJobRunState. // All options are ignored. -func (r *Repository) UpdateProgress(ctx context.Context, runId string, completed, total int, _ ...Option) (*Run, error) { +func (r *Repository) UpdateProgress(ctx context.Context, runId string, completed, total, retries int, _ ...Option) (*Run, error) { const op = "job.(Repository).UpdateProgress" if runId == "" { return nil, errors.New(ctx, errors.InvalidParameter, op, "missing run id") @@ -81,7 +81,7 @@ func (r *Repository) UpdateProgress(ctx context.Context, runId string, completed run.PrivateId = runId _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { - rows, err := w.Query(ctx, updateProgressQuery, []any{completed, total, runId}) + rows, err := w.Query(ctx, updateProgressQuery, []any{completed, total, retries, runId}) if err != nil { return errors.Wrap(ctx, err, op) } @@ -134,7 +134,7 @@ func (r *Repository) UpdateProgress(ctx context.Context, runId string, completed // or interrupted), any future calls to CompleteRun will return an error with Code // errors.InvalidJobRunState. // All options are ignored. -func (r *Repository) CompleteRun(ctx context.Context, runId string, nextRunIn time.Duration, completed, total int, _ ...Option) (*Run, error) { +func (r *Repository) CompleteRun(ctx context.Context, runId string, nextRunIn time.Duration, completed, total, retries int, _ ...Option) (*Run, error) { const op = "job.(Repository).CompleteRun" if runId == "" { return nil, errors.New(ctx, errors.InvalidParameter, op, "missing run id") @@ -148,7 +148,7 @@ func (r *Repository) CompleteRun(ctx context.Context, runId string, nextRunIn ti // persisted by the scheduler's monitor jobs loop. // Add an on update sql trigger to protect the job_run table, once progress // values are used in the critical path. - rows, err := w.Query(ctx, completeRunQuery, []any{completed, total, runId}) + rows, err := w.Query(ctx, completeRunQuery, []any{completed, total, retries, runId}) if err != nil { return errors.Wrap(ctx, err, op) } @@ -220,7 +220,7 @@ func (r *Repository) CompleteRun(ctx context.Context, runId string, nextRunIn ti // or interrupted), any future calls to FailRun will return an error with Code // errors.InvalidJobRunState. // All options are ignored. -func (r *Repository) FailRun(ctx context.Context, runId string, completed, total int, _ ...Option) (*Run, error) { +func (r *Repository) FailRun(ctx context.Context, runId string, completed, total, retries int, _ ...Option) (*Run, error) { const op = "job.(Repository).FailRun" if runId == "" { return nil, errors.New(ctx, errors.InvalidParameter, op, "missing run id") @@ -234,7 +234,7 @@ func (r *Repository) FailRun(ctx context.Context, runId string, completed, total // persisted by the scheduler's monitor jobs loop. // Add an on update sql trigger to protect the job_run table, once progress // values are used in the critical path. - rows, err := w.Query(ctx, failRunQuery, []any{completed, total, runId}) + rows, err := w.Query(ctx, failRunQuery, []any{completed, total, retries, runId}) if err != nil { return errors.Wrap(ctx, err, op) } diff --git a/internal/scheduler/job/repository_run_test.go b/internal/scheduler/job/repository_run_test.go index a468535dc..fcc415e25 100644 --- a/internal/scheduler/job/repository_run_test.go +++ b/internal/scheduler/job/repository_run_test.go @@ -216,7 +216,7 @@ func TestRepository_RunJobsOrder(t *testing.T) { assert.Equal(run.JobPluginId, firstJob.PluginId) // End first job with time between last and middle - _, err = repo.CompleteRun(ctx, run.PrivateId, -6*time.Hour, 0, 0) + _, err = repo.CompleteRun(ctx, run.PrivateId, -6*time.Hour, 0, 0, 0) require.NoError(err) runs, err = repo.RunJobs(ctx, server.PrivateId) @@ -260,7 +260,7 @@ func TestRepository_UpdateProgress(t *testing.T) { job := testJob(t, conn, "name", "description", wrapper) type args struct { - completed, total int + completed, total, retries int } tests := []struct { @@ -342,10 +342,12 @@ func TestRepository_UpdateProgress(t *testing.T) { }, }, args: args{ - total: 10, + total: 10, + retries: 1, }, want: args{ - total: 10, + total: 10, + retries: 1, }, }, { @@ -381,10 +383,12 @@ func TestRepository_UpdateProgress(t *testing.T) { args: args{ completed: 10, total: 20, + retries: 1, }, want: args{ completed: 10, total: 20, + retries: 1, }, }, { @@ -422,7 +426,7 @@ func TestRepository_UpdateProgress(t *testing.T) { privateId = tt.orig.PrivateId } - got, err := repo.UpdateProgress(ctx, privateId, tt.args.completed, tt.args.total) + got, err := repo.UpdateProgress(ctx, privateId, tt.args.completed, tt.args.total, tt.args.retries) if tt.wantErr { require.Error(err) assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err) @@ -455,7 +459,7 @@ func TestRepository_UpdateProgress(t *testing.T) { require.NoError(err) require.NotNil(repo) - got, err := repo.UpdateProgress(ctx, "fake-run-id", 0, 0) + got, err := repo.UpdateProgress(ctx, "fake-run-id", 0, 0, 0) require.Error(err) require.Nil(got) assert.Truef(errors.Match(errors.T(errors.RecordNotFound), err), "Unexpected error %s", err) @@ -476,7 +480,7 @@ func TestRepository_CompleteRun(t *testing.T) { job := testJob(t, conn, "name", "description", wrapper) type args struct { - completed, total int + completed, total, retries int } tests := []struct { name string @@ -557,7 +561,7 @@ func TestRepository_CompleteRun(t *testing.T) { Status: Running.string(), }, }, - args: args{completed: 10, total: 20}, + args: args{completed: 10, total: 20, retries: 1}, }, } @@ -577,7 +581,7 @@ func TestRepository_CompleteRun(t *testing.T) { privateId = tt.orig.PrivateId } - got, err := repo.CompleteRun(ctx, privateId, tt.nextRunIn, tt.args.completed, tt.args.total) + got, err := repo.CompleteRun(ctx, privateId, tt.nextRunIn, tt.args.completed, tt.args.total, tt.args.retries) if tt.wantErr { require.Error(err) assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err) @@ -597,6 +601,7 @@ func TestRepository_CompleteRun(t *testing.T) { assert.Equal(Completed.string(), got.Status) assert.Equal(tt.args.completed, int(got.CompletedCount)) assert.Equal(tt.args.total, int(got.TotalCount)) + assert.Equal(tt.args.retries, int(got.RetriesCount)) updatedJob, err := repo.LookupJob(ctx, tt.orig.JobName) assert.NoError(err) @@ -621,7 +626,7 @@ func TestRepository_CompleteRun(t *testing.T) { require.NoError(err) require.NotNil(repo) - got, err := repo.CompleteRun(ctx, "fake-run-id", time.Hour, 0, 0) + got, err := repo.CompleteRun(ctx, "fake-run-id", time.Hour, 0, 0, 0) require.Error(err) require.Nil(got) assert.Truef(errors.Match(errors.T(errors.RecordNotFound), err), "Unexpected error %s", err) @@ -642,7 +647,7 @@ func TestRepository_FailRun(t *testing.T) { job := testJob(t, conn, "name", "description", wrapper) type args struct { - completed, total int + completed, total, retries int } tests := []struct { name string @@ -721,7 +726,7 @@ func TestRepository_FailRun(t *testing.T) { Status: Running.string(), }, }, - args: args{completed: 10, total: 20}, + args: args{completed: 10, total: 20, retries: 5}, }, } @@ -741,7 +746,7 @@ func TestRepository_FailRun(t *testing.T) { privateId = tt.orig.PrivateId } - got, err := repo.FailRun(ctx, privateId, tt.args.completed, tt.args.total) + got, err := repo.FailRun(ctx, privateId, tt.args.completed, tt.args.total, tt.args.retries) if tt.wantErr { require.Error(err) assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err) @@ -761,6 +766,7 @@ func TestRepository_FailRun(t *testing.T) { assert.Equal(Failed.string(), got.Status) assert.Equal(tt.args.completed, int(got.CompletedCount)) assert.Equal(tt.args.total, int(got.TotalCount)) + assert.Equal(tt.args.retries, int(got.RetriesCount)) // Delete job run so it does not clash with future runs _, err = repo.deleteRun(context.Background(), privateId) @@ -774,7 +780,7 @@ func TestRepository_FailRun(t *testing.T) { require.NoError(err) require.NotNil(repo) - got, err := repo.FailRun(ctx, "fake-run-id", 0, 0) + got, err := repo.FailRun(ctx, "fake-run-id", 0, 0, 0) require.Error(err) require.Nil(got) assert.Truef(errors.Match(errors.T(errors.RecordNotFound), err), "Unexpected error %s", err) diff --git a/internal/scheduler/job/store/job.pb.go b/internal/scheduler/job/store/job.pb.go index 7bc6229cf..c6d4020a8 100644 --- a/internal/scheduler/job/store/job.pb.go +++ b/internal/scheduler/job/store/job.pb.go @@ -134,6 +134,10 @@ type JobRun struct { // total_count is set during an update to indicate the progress goal. // @inject_tag: `gorm:"default:0"` TotalCount uint32 `protobuf:"varint,9,opt,name=total_count,json=totalCount,proto3" json:"total_count,omitempty" gorm:"default:0"` + // retries_count is set during an update to indicate the number of times + // a job has retried work. + // @inject_tag: `gorm:"default:0"` + RetriesCount uint32 `protobuf:"varint,12,opt,name=retries_count,json=retriesCount,proto3" json:"retries_count,omitempty" gorm:"default:0"` // status of the job run (running, completed, failed or interrupted). // @inject_tag: `gorm:"not_null"` Status string `protobuf:"bytes,10,opt,name=status,proto3" json:"status,omitempty" gorm:"not_null"` @@ -230,6 +234,13 @@ func (x *JobRun) GetTotalCount() uint32 { return 0 } +func (x *JobRun) GetRetriesCount() uint32 { + if x != nil { + return x.RetriesCount + } + return 0 +} + func (x *JobRun) GetStatus() string { if x != nil { return x.Status @@ -266,7 +277,7 @@ var file_controller_storage_job_store_v1_job_proto_rawDesc = []byte{ 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x10, 0x6e, 0x65, 0x78, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x64, 0x52, - 0x75, 0x6e, 0x22, 0xd9, 0x03, 0x0a, 0x06, 0x4a, 0x6f, 0x62, 0x52, 0x75, 0x6e, 0x12, 0x1d, 0x0a, + 0x75, 0x6e, 0x22, 0xfe, 0x03, 0x0a, 0x06, 0x4a, 0x6f, 0x62, 0x52, 0x75, 0x6e, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x49, 0x64, 0x12, 0x22, 0x0a, 0x0d, 0x6a, 0x6f, 0x62, 0x5f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, @@ -291,16 +302,18 @@ var file_controller_storage_job_store_v1_job_proto_rawDesc = []byte{ 0x08, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e, 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1f, 0x0a, 0x0b, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x74, 0x6f, 0x74, 0x61, - 0x6c, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, - 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x23, - 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, - 0x0b, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, - 0x72, 0x49, 0x64, 0x52, 0x09, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x42, 0x42, - 0x5a, 0x40, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, - 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, - 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, - 0x65, 0x72, 0x2f, 0x6a, 0x6f, 0x62, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, - 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x6c, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x23, 0x0a, 0x0d, 0x72, 0x65, 0x74, 0x72, 0x69, 0x65, + 0x73, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0c, 0x72, + 0x65, 0x74, 0x72, 0x69, 0x65, 0x73, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, + 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x61, + 0x74, 0x75, 0x73, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, + 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x74, + 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x49, 0x64, 0x52, 0x09, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, + 0x5f, 0x69, 0x64, 0x42, 0x42, 0x5a, 0x40, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, + 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, + 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x63, + 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x2f, 0x6a, 0x6f, 0x62, 0x2f, 0x73, 0x74, 0x6f, 0x72, + 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index afea3c905..b71f77b93 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -228,7 +228,7 @@ func (s *Scheduler) schedule(ctx context.Context, wg *sync.WaitGroup) { err := s.runJob(ctx, wg, r) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error starting job")) - if _, inner := repo.FailRun(ctx, r.PrivateId, 0, 0); inner != nil { + if _, inner := repo.FailRun(ctx, r.PrivateId, 0, 0, 0); inner != nil { event.WriteError(ctx, op, inner, event.WithInfoMsg("error updating failed job run")) } } @@ -273,10 +273,10 @@ func (s *Scheduler) runJob(ctx context.Context, wg *sync.WaitGroup, r *job.Run) if inner != nil { event.WriteError(ctx, op, inner, event.WithInfoMsg("error getting next run time", "name", j.Name())) } - _, updateErr = repo.CompleteRun(ctx, r.PrivateId, nextRun, status.Completed, status.Total) + _, updateErr = repo.CompleteRun(ctx, r.PrivateId, nextRun, status.Completed, status.Total, status.Retries) default: event.WriteError(ctx, op, runErr, event.WithInfoMsg("job run failed", "run id", r.PrivateId, "name", j.Name())) - _, updateErr = repo.FailRun(ctx, r.PrivateId, status.Completed, status.Total) + _, updateErr = repo.FailRun(ctx, r.PrivateId, status.Completed, status.Total, status.Retries) } if updateErr != nil { @@ -333,7 +333,7 @@ func (s *Scheduler) updateRunningJobProgress(ctx context.Context, j *runningJob) return fmt.Errorf("error creating job repo %w", err) } status := j.status() - _, err = repo.UpdateProgress(ctx, j.runId, status.Completed, status.Total) + _, err = repo.UpdateProgress(ctx, j.runId, status.Completed, status.Total, status.Retries) if errors.Match(errors.T(errors.InvalidJobRunState), err) { // Job has been persisted with a final run status, cancel job context to trigger early exit. j.cancelCtx()