jobs: retry jobs with exponential backoff

[sajjadrizvi]

This commit adds a mechanism to retry jobs with exponentially increasing
delays. This is achieved through two new columns in system.jobs table,
last_run and num_runs. In addition, this commit adds cluster settings
to control exponential-backoff parameters, initial delay and max delay,
with corresponding settings jobs.registry.retry.initial_delay and
jobs.registry.retry.max_delay. Finally, this commit adds a new
partial-index in the jobs table that improves the performance of periodic
queries run by registry in each node.

Release note (general change): The behavior for retrying jobs, which fail
due to a retriable error or due to job coordinator failure, is now delayed
using exponential backoff. Before this change, jobs which failed in a
retryable manner, would be resumed immediately on a different coordinator.
This change reduces the impact of recurrently failing jobs on the cluster.
This change adds two new cluster settings that control this behavior:
"jobs.registry.retry.initial_delay" and "jobs.registry.retry.max_delay",
which respectively control initial delay and maximum delay between
resumptions.

Fixes #44594
Fixes #65080

[cockroach-teamcity]

This change is 

[ajwerner]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @sajjadrizvi)

---

pkg/jobs/adopt.go, line 51 at r3 (raw file):

      SET claim_session_id = $1, claim_instance_id = $2
    WHERE ((claim_session_id IS NULL)
      AND (status IN ` + claimableStatusTupleString + `))

Why these extra parens?

---

pkg/jobs/adopt.go, line 86 at r3 (raw file):

	// Select only those jobs that have their next retry time before the current time.
	// 2^62 is the max positive number.
	// $5 is the base in exponential backoff calculation, and $6 is the max retry delay.

I think your placeholders are off now. Also, $6 isn't the "base" the way we usually talk about bases in exponentials but rather the "starting value".

---

pkg/jobs/config.go, line 112 at r3 (raw file):

	retryBaseSetting = settings.RegisterDurationSetting(
		retryBaseSettingKey,
		"the base multiplier to calculate the exponential-backoff delay "+

nit: starting value. You could also supply a base if you wanted (it should be positive). Right now it's 2.

---

pkg/migration/migrations/retry_jobs_with_exponential_backoff.go, line 24 at r3 (raw file):

	ctx context.Context, _ clusterversion.ClusterVersion, d migration.TenantDeps,
) error {
	const query = `

The problem with this approach is that it might return before the schema change is done. Imagine that we start this migration and issue this transaction. It will create the job to do the schema change. Now, while the schema change is running, we crash. When we come back around, the schema change will still be running but the below commands will no-op and we'll return happily. This is not a good place to be. We need to make sure that the table really is changed.

There are a bunch of ways to do this. One way is to check to make sure that the jobs table does not have any running jobs and if it does, to wait for them. You could do that at the top of the loop. There are two basic approaches you could take to that task:

1. Use a descs.Txn to read the jobs table and check for the new columns or for a pending job
2. Use sql to do the same thing by using crdb_internal.pb_to_json

I don't have a strong preference.

---

pkg/migration/migrations/retry_jobs_with_exponential_backoff.go, line 26 at r3 (raw file):

	const query = `
BEGIN;
	ALTER TABLE IF EXISTS system.jobs

The table will definitely exist so nix the first IF EXISTS.

---

pkg/sql/catalog/systemschema/system.go, line 897 at r3 (raw file):

				StoreColumnNames:    []string{"claim_instance_id", "num_runs", "last_run"},
				StoreColumnIDs:      []descpb.ColumnID{9, 10, 11},
				KeySuffixColumnIDs:  []descpb.ColumnID{1}, //TODO(sajjad): To discuss: What should be the values here?

The values should be all of the columns of the primary key which are not in the KeyColumnIDs. In this case, 1.

[ajwerner]

We wanted to create another index for two reasons. The main one is to make it a partial index. The other reason is to allow us to order it on claim_session_id

[ajwerner]

Speaking of which, this is missing the predicate.

[ajwerner]

I'm not so sure about these IF NOT EXISTS

[ajwerner]

Be careful about aligning the naming on the index.

[sajjadrizvi]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ajwerner)

---

pkg/jobs/adopt.go, line 51 at r3 (raw file):

Previously, ajwerner wrote…

Why these extra parens?

This is because when we intercept the query in the StatementFilter testing knob, the statement has these extra params even if we don't add them. As we compare the received statement with this const in TestRegistrySettingUpdate, the statements don't match if we don't add these extra params.

---

pkg/jobs/adopt.go, line 86 at r3 (raw file):

Previously, ajwerner wrote…

I think your placeholders are off now. Also, $6 isn't the "base" the way we usually talk about bases in exponentials but rather the "starting value".

Done.

---

pkg/migration/migrations/retry_jobs_with_exponential_backoff.go, line 24 at r3 (raw file):

Previously, ajwerner wrote…

The problem with this approach is that it might return before the schema change is done. Imagine that we start this migration and issue this transaction. It will create the job to do the schema change. Now, while the schema change is running, we crash. When we come back around, the schema change will still be running but the below commands will no-op and we'll return happily. This is not a good place to be. We need to make sure that the table really is changed.

There are a bunch of ways to do this. One way is to check to make sure that the jobs table does not have any running jobs and if it does, to wait for them. You could do that at the top of the loop. There are two basic approaches you could take to that task:

1. Use a descs.Txn to read the jobs table and check for the new columns or for a pending job
2. Use sql to do the same thing by using crdb_internal.pb_to_json

I don't have a strong preference.

As we sketched out together, we have taken the first approach.

---

pkg/migration/migrations/retry_jobs_with_exponential_backoff.go, line 40 at r5 (raw file):

Previously, ajwerner wrote…

Be careful about aligning the naming on the index.

Do you mean the ordering of the columns based on the schema in system.go? I didn't order them correctly.
I added IF NOT EXISTS to cater for transient failures that may happen after the index has been created but before completing the migration.

---

pkg/sql/catalog/systemschema/system.go, line 961 at r5 (raw file):

Previously, ajwerner wrote…

Speaking of which, this is missing the predicate.

Thanks for pointing out. I have added the predicate.

[ajwerner]

func equalMessage(a, b proto.Message) error {
    aBytes, err := protoutil.Marshal(a)
    if err != nil { ... }
    bBytes, err := protoutil.Marshal(b)
    if err != nil { ... }
    if bytes.Equal(aBytes, bBytes) { return nil }
    return errors.Errorf("mismatch for messages of types %T&%T: %s", a, b, strings.Join(pretty.Diff(a, b), "\n")
}

[sajjadrizvi]

TFTR.

bors r=ajwerner

[craig]

This PR was included in a batch that was canceled, it will be automatically retried

[craig]

Build failed (retrying...):

• GitHub CI (Cockroach)

[craig]

Build succeeded:

• GitHub CI (Cockroach)

[andreimatei]

This patch made TestFailureToMarkCanceledReversalLeadsToCanceledStatus take 2m (from being very fast). Perhaps other tests slowed down too?
Do you mind seeing if that's expected / if anything can be done to the test and, if not, at the very least skip.UnderShort() the test?

[ajwerner]

This patch made TestFailureToMarkCanceledReversalLeadsToCanceledStatus take 2m (from being very fast). Perhaps other tests slowed down too?
Do you mind seeing if that's expected / if anything can be done to the test and, if not, at the very least skip.UnderShort() the test?

Thanks for the heads up! #69103

[andreimatei]

Thanks! Maybe check that no other tests slowed similarly by comparing package run times; I haven't done that.