restore: prevent retry loop after failed full cluster online restore

Previously, if online restore failed on a full cluster restore, it would
get stuck in a retry loop in its attempt to set all descriptors offline.
Those descriptors would include the temporary system tables, which were
dropped at the end of the link phase. This patch updates online restore
recovery to ignore these dropped descriptors instead of failing.

Epic: CRDB-51394

Fixes: #148088

Release note: Online restore no longer gets stuck in a retry loop when
reverting after a failed or canceled download job.

Author: kev-cao

pkg/backup/restore_online.go

@@ -892,11 +892,11 @@ func setDescriptorsOffline(
 		return nil
 	}
 
-	for i := range details.TableDescs {
-		mutableTable, err := descCol.MutableByID(txn.KV()).Table(ctx, details.TableDescs[i].ID)
-		if err != nil {
-			return err
-		}
+	mutableTables, err := getUndroppedTablesFromRestore(ctx, txn.KV(), details, descCol)
+	if err != nil {
+		return errors.Wrapf(err, "set descriptors offline: getting undropped tables from restore")
+	}
+	for _, mutableTable := range mutableTables {
 		if err := writeDesc(mutableTable); err != nil {
 			return err
 		}

pkg/backup/restore_online_test.go

@@ -204,12 +204,73 @@ func TestOnlineRestoreRecovery(t *testing.T) {
 	})
 }
 
+// We run full cluster online restore recovery in a separate environment since
+// it requires dropping all databases and will impact other tests.
+func TestFullClusterOnlineRestoreRecovery(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	tmpDir := t.TempDir()
+
+	defer nodelocal.ReplaceNodeLocalForTesting(tmpDir)()
+
+	const numAccounts = 1000
+
+	externalStorage := "nodelocal://1/backup"
+
+	_, sqlDB, _, cleanupFn := backupRestoreTestSetupWithParams(t, singleNode, numAccounts, InitManualReplication, orParams)
+	defer cleanupFn()
+
+	sqlDB.Exec(t, "SET CLUSTER SETTING jobs.debug.pausepoints = 'restore.before_download'")
+	sqlDB.Exec(t, fmt.Sprintf("BACKUP INTO '%s'", externalStorage))
+
+	// Reset cluster for full cluster restore.
+	dbs := sqlDB.QueryStr(
+		t, "SELECT database_name FROM [SHOW DATABASES] WHERE database_name != 'system'",
+	)
+	sqlDB.Exec(t, "USE system")
+	for _, db := range dbs {
+		sqlDB.Exec(t, fmt.Sprintf("DROP DATABASE %s CASCADE", db[0]))
+	}
+
+	var linkJobID jobspb.JobID
+	sqlDB.QueryRow(
+		t,
+		fmt.Sprintf(
+			"RESTORE FROM LATEST IN '%s' WITH EXPERIMENTAL DEFERRED COPY, detached", externalStorage,
+		),
+	).Scan(&linkJobID)
+	jobutils.WaitForJobToSucceed(t, sqlDB, linkJobID)
+	var downloadJobID jobspb.JobID
+	sqlDB.QueryRow(t, latestDownloadJobIDQuery).Scan(&downloadJobID)
+	jobutils.WaitForJobToPause(t, sqlDB, downloadJobID)
+	corruptBackup(t, sqlDB, tmpDir, externalStorage)
+	sqlDB.Exec(t, "SET CLUSTER SETTING jobs.debug.pausepoints = ''")
+	sqlDB.Exec(t, fmt.Sprintf("RESUME JOB %d", downloadJobID))
+	jobutils.WaitForJobToFail(t, sqlDB, downloadJobID)
+}
+
 func corruptBackup(t *testing.T, sqlDB *sqlutils.SQLRunner, ioDir string, uri string) {
-	var filePath string
-	filePathQuery := fmt.Sprintf("SELECT path FROM [SHOW BACKUP FILES FROM LATEST IN '%s'] LIMIT 1", uri)
-	parsedURI, err := url.Parse(strings.Replace(uri, "'", "", -1))
-	sqlDB.QueryRow(t, filePathQuery).Scan(&filePath)
+	var filePath, spanStart, spanEnd string
+	// We delete the last SST file in SHOW BACKUP to ensure the deletion of an SST
+	// file that backs a user-table.
+	// https://github.com/cockroachdb/cockroach/issues/148408 illustrates how
+	// deleting the backing SST file of a system table will not necessarily cause
+	// a download job to fail.
+	filePathQuery := fmt.Sprintf(
+		`SELECT path, start_pretty, end_pretty FROM
+		(
+			SELECT row_number() OVER (), *
+			FROM [SHOW BACKUP FILES FROM LATEST IN '%s']
+		)
+		ORDER BY row_number DESC
+		LIMIT 1`,
+		uri,
+	)
+	parsedURI, err := url.Parse(strings.ReplaceAll(uri, "'", ""))
+	sqlDB.QueryRow(t, filePathQuery).Scan(&filePath, &spanStart, &spanEnd)
 	fullPath := filepath.Join(ioDir, parsedURI.Path, filePath)
+	t.Logf("deleting backup file %s covering span [%s, %s)", fullPath, spanStart, spanEnd)
 	require.NoError(t, err)
 	require.NoError(t, os.Remove(fullPath))
 }