restore: mark pebble corruption errors on backup file reads as failures

This patch ensures that regular restore will fast fail when it reads a
corrupted backup file.

Fixes #150258

Release note: none

Author: msbutler

pkg/backup/BUILD.bazel

@@ -159,6 +159,7 @@ go_library(
         "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
+        "@com_github_cockroachdb_pebble//:pebble",
         "@com_github_cockroachdb_redact//:redact",
         "@com_github_gogo_protobuf//types",
         "@com_github_robfig_cron_v3//:cron",

pkg/backup/restore_data_processor.go

@@ -454,6 +454,8 @@ func (rd *restoreDataProcessor) runRestoreWorkers(
 	})
 }
 
+var backupFileReadError = errors.New("error reading backup file")
+
 func (rd *restoreDataProcessor) processRestoreSpanEntry(
 	ctx context.Context, kr *KeyRewriter, sst mergedSST,
 ) (kvpb.BulkOpSummary, error) {
@@ -537,9 +539,8 @@ func (rd *restoreDataProcessor) processRestoreSpanEntry(
 	for iter.SeekGE(startKeyMVCC); ; iter.NextKey() {
 		ok, err := iter.Valid()
 		if err != nil {
-			return summary, err
+			return summary, errors.Join(backupFileReadError, err)
 		}
-
 		if !ok {
 			if verbose {
 				log.Infof(ctx, "iterator exhausted")
@@ -560,12 +561,12 @@ func (rd *restoreDataProcessor) processRestoreSpanEntry(
 
 		v, err := iter.UnsafeValue()
 		if err != nil {
-			return summary, err
+			return summary, errors.Join(backupFileReadError, err)
 		}
 		valueScratch = append(valueScratch[:0], v...)
 		value, err := storage.DecodeValueFromMVCCValue(valueScratch)
 		if err != nil {
-			return summary, err
+			return summary, errors.Join(backupFileReadError, err)
 		}
 
 		key.Key, ok, err = kr.RewriteKey(key.Key, key.Timestamp.WallTime)

pkg/backup/restore_job.go

@@ -79,6 +79,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/pebble"
 )
 
 var (
@@ -215,6 +216,10 @@ func restoreWithRetry(
 		if errors.HasType(err, &kvpb.InsufficientSpaceError{}) {
 			return roachpb.RowCount{}, jobs.MarkPauseRequestError(errors.UnwrapAll(err))
 		}
+
+		if pebble.IsCorruptionError(err) && errors.Is(err, backupFileReadError) {
+			return roachpb.RowCount{}, jobs.MarkAsPermanentJobError(err)
+		}
 		// If we are draining, it is unlikely we can start a
 		// new DistSQL flow. Exit with a retryable error so
 		// that another node can pick up the job.