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()