Merge pull request #151080 from dt/backport25.3-151058

release-25.3: backup: fix external storage error propagation

Author: msbutler

pkg/backup/backup_processor.go

@@ -292,6 +292,14 @@ type spanAndTime struct {
 	finishesSpec bool
 }
 
+type errInjectingStorage struct {
+	cloud.ExternalStorage
+}
+
+func (e errInjectingStorage) Writer(_ context.Context, _ string) (io.WriteCloser, error) {
+	return nil, errors.New("injected error")
+}
+
 func runBackupProcessor(
 	ctx context.Context,
 	flowCtx *execinfra.FlowCtx,
@@ -393,12 +401,19 @@ func runBackupProcessor(
 		Settings:  &flowCtx.Cfg.Settings.SV,
 		ElideMode: spec.ElidePrefix,
 	}
+
 	storage, err := flowCtx.Cfg.ExternalStorage(ctx, dest, cloud.WithClientName("backup"))
 	if err != nil {
 		return err
 	}
 	defer logClose(ctx, storage, "external storage")
 
+	if backupKnobs, ok := flowCtx.TestingKnobs().BackupRestoreTestingKnobs.(*sql.BackupRestoreTestingKnobs); ok {
+		if fn := backupKnobs.InjectErrorsInBackupRowDataStorage; fn != nil && fn() {
+			storage = errInjectingStorage{storage}
+		}
+	}
+
 	// Start start a group of goroutines which each pull spans off of `todo` and
 	// send export requests. Any spans that encounter lock conflict errors during
 	// Export are put back on the todo queue for later processing.
@@ -683,7 +698,7 @@ func runBackupProcessor(
 							var writeErr error
 							resumeSpan.span.Key, writeErr = sink.Write(ctx, ret)
 							if writeErr != nil {
-								return err
+								return writeErr
 							}
 						}
 						// Emit the stats for the processed ExportRequest.

pkg/backup/backup_test.go

@@ -6321,6 +6321,36 @@ func TestPublicIndexTableSpans(t *testing.T) {
 	}
 }
 
+// TestBackupStorageErrorPropagates ensures that errors from writing to storage
+// propagate correctly during a backup operation.
+func TestBackupStorageErrorPropagates(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	const numAccounts = 1000
+
+	var fail atomic.Bool
+
+	params := base.TestClusterArgs{}
+	knobs := base.TestingKnobs{
+		DistSQL: &execinfra.TestingKnobs{BackupRestoreTestingKnobs: &sql.BackupRestoreTestingKnobs{
+			InjectErrorsInBackupRowDataStorage: func() bool { return fail.Load() },
+		}},
+	}
+	params.ServerArgs.Knobs = knobs
+
+	tc, _, _, cleanupFn := backupRestoreTestSetupWithParams(t, singleNode, numAccounts, InitManualReplication, params)
+	defer cleanupFn()
+	db := tc.ServerConn(0)
+	runner := sqlutils.MakeSQLRunner(db)
+
+	runner.Exec(t, "BACKUP DATABASE data INTO 'nodelocal://1/success'")
+	runner.Exec(t, "RESTORE DATABASE data FROM LATEST IN 'nodelocal://1/success' WITH new_db_name = 'restored'")
+	runner.CheckQueryResults(t, "SELECT count(*) FROM restored.bank", [][]string{{"1000"}})
+	fail.Store(true)
+	runner.ExpectErr(t, "injected", "BACKUP DATABASE data INTO 'nodelocal://1/failure'")
+}
+
 // TestRestoreJobErrorPropagates ensures that errors from creating the job
 // record propagate correctly.
 func TestRestoreErrorPropagates(t *testing.T) {

pkg/sql/exec_util_backup.go

@@ -47,6 +47,11 @@ type BackupRestoreTestingKnobs struct {
 	// span has been exported.
 	RunAfterExportingSpanEntry func(ctx context.Context, response *kvpb.ExportResponse)
 
+	// InjectErrorsInBackupRowDataStorage, if non-nil and returning true, causes
+	// errors to be injected when backup processors attempts to write row data to
+	// the external storage.
+	InjectErrorsInBackupRowDataStorage func() bool
+
 	// BackupMonitor is used to overwrite the monitor used by backup during
 	// testing. This is typically the bulk mem monitor if not
 	// specified here.