release-21.1: sql/gcjob: retry failed GC jobs

pkg/sql/gcjob/gc_job.go

@@ -80,7 +80,12 @@ func performGC(
 }
 
 // Resume is part of the jobs.Resumer interface.
-func (r schemaChangeGCResumer) Resume(ctx context.Context, execCtx interface{}) error {
+func (r schemaChangeGCResumer) Resume(ctx context.Context, execCtx interface{}) (err error) {
+	defer func() {
+		if err != nil && !r.isPermanentGCError(err) {
+			err = errors.Mark(err, jobs.NewRetryJobError("gc"))
+		}
+	}()
 	p := execCtx.(sql.JobExecContext)
 	// TODO(pbardea): Wait for no versions.
 	execCfg := p.ExecCfg()
@@ -181,6 +186,15 @@ func (r schemaChangeGCResumer) OnFailOrCancel(context.Context, interface{}) erro
 	return nil
 }
 
+// isPermanentGCError returns true if the error is a permanent job failure,
+// which indicates that the failed GC job cannot be retried.
+func (r *schemaChangeGCResumer) isPermanentGCError(err error) bool {
+	// Currently we classify errors based on Schema Change function to backport
+	// to 20.2 and 21.1. This functionality should be changed once #44594 is
+	// implemented.
+	return sql.IsPermanentSchemaChangeError(err)
+}
+
 func init() {
 	createResumerFn := func(job *jobs.Job, settings *cluster.Settings) jobs.Resumer {
 		return &schemaChangeGCResumer{

pkg/sql/gcjob_test/BUILD.bazel

@@ -13,6 +13,8 @@ go_test(
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv",
+        "//pkg/kv/kvserver",
+        "//pkg/roachpb",
         "//pkg/security",
         "//pkg/security/securitytest",
         "//pkg/server",
@@ -27,6 +29,7 @@ go_test(
         "//pkg/testutils/serverutils",
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
+        "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/randutil",

pkg/sql/gcjob_test/gc_job_test.go

@@ -14,6 +14,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -22,6 +23,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkeys"
@@ -34,6 +37,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
@@ -415,3 +419,45 @@ func TestGCResumer(t *testing.T) {
 		require.Error(t, sj.AwaitCompletion(ctx))
 	})
 }
+
+func TestGCJobRetry(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	var failed atomic.Value
+	failed.Store(false)
+	params := base.TestServerArgs{}
+	params.Knobs.Store = &kvserver.StoreTestingKnobs{
+		TestingRequestFilter: func(ctx context.Context, request roachpb.BatchRequest) *roachpb.Error {
+			_, ok := request.GetArg(roachpb.ClearRange)
+			if !ok {
+				return nil
+			}
+			if failed.Load().(bool) {
+				return nil
+			}
+			failed.Store(true)
+			return roachpb.NewError(&roachpb.BatchTimestampBeforeGCError{
+				Timestamp: hlc.Timestamp{},
+				Threshold: hlc.Timestamp{},
+			})
+		},
+	}
+	s, db, _ := serverutils.StartServer(t, params)
+	defer s.Stopper().Stop(ctx)
+	tdb := sqlutils.MakeSQLRunner(db)
+	tdb.Exec(t, "CREATE TABLE foo (i INT PRIMARY KEY)")
+	tdb.Exec(t, "ALTER TABLE foo CONFIGURE ZONE USING gc.ttlseconds = 1;")
+	tdb.Exec(t, "DROP TABLE foo CASCADE;")
+	var jobID int64
+	tdb.QueryRow(t, `
+SELECT job_id
+  FROM [SHOW JOBS]
+ WHERE job_type = 'SCHEMA CHANGE GC' AND description LIKE '%foo%';`,
+	).Scan(&jobID)
+	var status jobs.Status
+	tdb.QueryRow(t,
+		"SELECT status FROM [SHOW JOB WHEN COMPLETE $1]", jobID,
+	).Scan(&status)
+	require.Equal(t, jobs.StatusSucceeded, status)
+}

pkg/sql/schema_changer.go

@@ -138,11 +138,11 @@ func NewSchemaChangerForTesting(
 	}
 }
 
-// isPermanentSchemaChangeError returns true if the error results in
+// IsPermanentSchemaChangeError returns true if the error results in
 // a permanent failure of a schema change. This function is a allowlist
 // instead of a blocklist: only known safe errors are confirmed to not be
 // permanent errors. Anything unknown is assumed to be permanent.
-func isPermanentSchemaChangeError(err error) bool {
+func IsPermanentSchemaChangeError(err error) bool {
 	if err == nil {
 		return false
 	}
@@ -2186,7 +2186,7 @@ func (r schemaChangeResumer) Resume(ctx context.Context, execCtx interface{}) er
 					descID, mutationID,
 				)
 				return nil
-			case !isPermanentSchemaChangeError(scErr):
+			case !IsPermanentSchemaChangeError(scErr):
 				// Check if the error is on a allowlist of errors we should retry on,
 				// including the schema change not having the first mutation in line.
 				log.Warningf(ctx, "error while running schema change, retrying: %v", scErr)
@@ -2351,7 +2351,7 @@ func (r schemaChangeResumer) OnFailOrCancel(ctx context.Context, execCtx interfa
 			// We check for this case so that we can just return the error without
 			// wrapping it in a retry error.
 			return rollbackErr
-		case !isPermanentSchemaChangeError(rollbackErr):
+		case !IsPermanentSchemaChangeError(rollbackErr):
 			// Check if the error is on a allowlist of errors we should retry on, and
 			// have the job registry retry.
 			return jobs.NewRetryJobError(rollbackErr.Error())

pkg/sql/type_change.go

@@ -795,7 +795,7 @@ func (t *typeSchemaChanger) execWithRetry(ctx context.Context) error {
 				t.typeID,
 			)
 			return nil
-		case !isPermanentSchemaChangeError(tcErr):
+		case !IsPermanentSchemaChangeError(tcErr):
 			// If this isn't a permanent error, then retry.
 			log.Infof(ctx, "retrying type schema change due to retriable error %v", tcErr)
 		default:
@@ -869,7 +869,7 @@ func (t *typeChangeResumer) OnFailOrCancel(ctx context.Context, execCtx interfac
 				"descriptor %d not found for type change job; assuming it was dropped, and exiting",
 				tc.typeID,
 			)
-		case !isPermanentSchemaChangeError(rollbackErr):
+		case !IsPermanentSchemaChangeError(rollbackErr):
 			return jobs.NewRetryJobError(rollbackErr.Error())
 		default:
 			return rollbackErr