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.

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;

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;

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}

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 {

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)
 

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

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

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)

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 (

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