MDEV-31851 After crash recovery, undo tablespace fails to open

Problem:
========
- InnoDB fails to open undo tablespace when page0 is corrupted
and fails to throw error.

Solution:
=========
- InnoDB throws DB_CORRUPTION error when InnoDB encounters
page0 corruption of undo tablespace.

- InnoDB restores the page0 of undo tablespace from
doublewrite buffer if it encounters page corruption

- Moved Datafile::restore_from_doublewrite() to
recv_dblwr_t::restore_first_page(). So that undo
tablespace and system tablespace can use this function
instead of duplicating the code

srv_undo_tablespace_open(): Returns 0 if file doesn't exist
or ULINT_UNDEFINED if page0 is corrupted.

mysql-test/suite/innodb/r/undo_space_dblwr.result

+call mtr.add_suppression("Checksum mismatch in the first page of file");
+show variables like 'innodb_doublewrite';
+Variable_name	Value
+innodb_doublewrite	ON
+create table t1(f1 int not null, f2 int not null)engine=innodb;
+insert into t1 values (1, 1);
+SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
+InnoDB		0 transactions not purged
+set GLOBAL innodb_log_checkpoint_now=1;
+# Make the first page dirty for undo tablespace
+set global innodb_saved_page_number_debug = 0;
+set global innodb_fil_make_page_dirty_debug = 1;
+SET GLOBAL innodb_max_dirty_pages_pct_lwm=0.0;
+SET GLOBAL innodb_max_dirty_pages_pct=0.0;
+# Kill the server
+# restart
+FOUND 1 /Checksum mismatch in the first page of file/ in mysqld.1.err
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+drop table t1;

mysql-test/suite/innodb/t/undo_space_dblwr.opt

+--innodb_undo_tablespaces=3
+--innodb_sys_tablespaces

mysql-test/suite/innodb/t/undo_space_dblwr.test

+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/not_embedded.inc
+call mtr.add_suppression("Checksum mismatch in the first page of file");
+let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
+let MYSQLD_DATADIR=`select @@datadir`;
+
+show variables like 'innodb_doublewrite';
+create table t1(f1 int not null, f2 int not null)engine=innodb;
+insert into t1 values (1, 1);
+
+SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
+--source include/wait_all_purged.inc
+
+set GLOBAL innodb_log_checkpoint_now=1;
+--source ../include/no_checkpoint_start.inc
+
+--echo # Make the first page dirty for undo tablespace
+set global innodb_saved_page_number_debug = 0;
+set global innodb_fil_make_page_dirty_debug = 1;
+
+SET GLOBAL innodb_max_dirty_pages_pct_lwm=0.0;
+SET GLOBAL innodb_max_dirty_pages_pct=0.0;
+
+sleep 1;
+--let CLEANUP_IF_CHECKPOINT=drop table t1;
+--source ../include/no_checkpoint_end.inc
+
+perl;
+use IO::Handle;
+my $fname= "$ENV{'MYSQLD_DATADIR'}/undo001";
+my $page_size = $ENV{INNODB_PAGE_SIZE};
+die unless open(FILE, "+<", $fname);
+sysread(FILE, $page, $page_size)==$page_size||die "Unable to read $name\n";
+substr($page, 49, 4) = pack("N", 1000);
+sysseek(FILE, 0, 0)||die "Unable to seek $fname\n";
+die unless syswrite(FILE, $page, $page_size) == $page_size;
+close FILE;
+EOF
+
+--source include/start_mysqld.inc
+let SEARCH_FILE= $MYSQLTEST_VARDIR/log/mysqld.1.err;
+let SEARCH_PATTERN= Checksum mismatch in the first page of file;
+--source include/search_pattern_in_file.inc
+
+check table t1;
+drop table t1;

storage/innobase/buf/buf0dblwr.cc

@@ -350,7 +350,7 @@ void buf_dblwr_t::recover()
   {
     byte *page= *i;
     const uint32_t page_no= page_get_page_no(page);
-    if (!page_no) /* recovered via Datafile::restore_from_doublewrite() */
+    if (!page_no) /* recovered via recv_dblwr_t::restore_first_page() */
       continue;
 
     const lsn_t lsn= mach_read_from_8(page + FIL_PAGE_LSN);

storage/innobase/fsp/fsp0file.cc

@@ -481,7 +481,8 @@ Datafile::validate_for_recovery()
 			return(err);
 		}
 
-		if (restore_from_doublewrite()) {
+		if (recv_sys.dblwr.restore_first_page(
+			m_space_id, m_filepath, m_handle)) {
 			return(DB_CORRUPTION);
 		}
 
@@ -768,61 +769,6 @@ Datafile::find_space_id()
 	return(DB_CORRUPTION);
 }
 
-
-/** Restore the first page of the tablespace from
-the double write buffer.
-@return whether the operation failed */
-bool
-Datafile::restore_from_doublewrite()
-{
-	if (srv_operation > SRV_OPERATION_EXPORT_RESTORED) {
-		return true;
-	}
-
-	/* Find if double write buffer contains page_no of given space id. */
-	const page_id_t	page_id(m_space_id, 0);
-	const byte*	page = recv_sys.dblwr.find_page(page_id);
-
-	if (!page) {
-		/* If the first page of the given user tablespace is not there
-		in the doublewrite buffer, then the recovery is going to fail
-		now. Hence this is treated as an error. */
-
-		ib::error()
-			<< "Corrupted page " << page_id
-			<< " of datafile '" << m_filepath
-			<< "' could not be found in the doublewrite buffer.";
-
-		return(true);
-	}
-
-	ulint	flags = mach_read_from_4(
-		FSP_HEADER_OFFSET + FSP_SPACE_FLAGS + page);
-
-	if (!fil_space_t::is_valid_flags(flags, m_space_id)) {
-		flags = fsp_flags_convert_from_101(flags);
-		/* recv_dblwr_t::validate_page() inside find_page()
-		checked this already. */
-		ut_ad(flags != ULINT_UNDEFINED);
-		/* The flags on the page should be converted later. */
-	}
-
-	ulint physical_size = fil_space_t::physical_size(flags);
-
-	ut_a(page_get_page_no(page) == page_id.page_no());
-
-	ib::info() << "Restoring page " << page_id
-		<< " of datafile '" << m_filepath
-		<< "' from the doublewrite buffer. Writing "
-		<< physical_size << " bytes into file '"
-		<< m_filepath << "'";
-
-	return(os_file_write(
-			IORequestWrite,
-			m_filepath, m_handle, page, 0, physical_size)
-	       != DB_SUCCESS);
-}
-
 /** Create a link filename based on the contents of m_name,
 open that file, and read the contents into m_filepath.
 @retval DB_SUCCESS if remote linked tablespace file is opened and read.

storage/innobase/fsp/fsp0sysspace.cc

@@ -594,7 +594,9 @@ SysTablespace::read_lsn_and_check_flags(lsn_t* flushed_lsn)
 
 		if (err != DB_SUCCESS
 		    && (retry == 1
-			|| it->restore_from_doublewrite())) {
+			|| recv_sys.dblwr.restore_first_page(
+				it->m_space_id, it->m_filepath,
+				it->handle()))) {
 
 			it->close();
 

storage/innobase/handler/ha_innodb.cc

@@ -18298,6 +18298,7 @@ checkpoint_now_set(THD*, st_mysql_sys_var*, void*, const void* save)
 		while (log_sys.last_checkpoint_lsn.load(
 			       std::memory_order_acquire)
 		       + SIZE_OF_FILE_CHECKPOINT
+		       + log_sys.framing_size()
 		       < (lsn= log_sys.get_lsn(std::memory_order_acquire))) {
 			log_make_checkpoint();
 			log_sys.log.flush();

storage/innobase/include/fsp0file.h

@@ -423,11 +423,6 @@ class Datafile {
 	else DB_ERROR. */
 	dberr_t find_space_id();
 
-	/** Restore the first page of the tablespace from
-	the double write buffer.
-	@return whether the operation failed */
-	bool restore_from_doublewrite();
-
 	/** Points into m_filepath to the file name with extension */
 	char*			m_filename;
 

storage/innobase/include/log0recv.h

@@ -134,6 +134,15 @@ struct recv_dblwr_t
   byte* find_page(const page_id_t page_id, const fil_space_t *space= NULL,
                   byte *tmp_buf= NULL);
 
+  /** Restore the first page of the given tablespace from
+  doublewrite buffer.
+  @param space_id  tablespace identifier
+  @param name      tablespace filepath
+  @param file      tablespace file handle
+  @return whether the operation failed */
+  bool restore_first_page(
+         ulint space_id, const char *name, os_file_t file);
+
   typedef std::deque<byte*, ut_allocator<byte*> > list;
 
   /** Recovered doublewrite buffer page frames */

storage/innobase/log/log0recv.cc

@@ -3819,6 +3819,11 @@ byte *recv_dblwr_t::find_page(const page_id_t page_id,
     if (page_get_page_no(page) != page_id.page_no() ||
         page_get_space_id(page) != page_id.space())
       continue;
+    uint32_t flags= mach_read_from_4(
+      FSP_HEADER_OFFSET + FSP_SPACE_FLAGS + page);
+    if (!fil_space_t::is_valid_flags(flags, page_id.space()))
+      continue;
+
     const lsn_t lsn= mach_read_from_8(page + FIL_PAGE_LSN);
     if (lsn <= max_lsn ||
         !validate_page(page_id, page, space, tmp_buf))
@@ -3827,9 +3832,38 @@ byte *recv_dblwr_t::find_page(const page_id_t page_id,
       memset(page + FIL_PAGE_LSN, 0, 8);
       continue;
     }
+
+    ut_a(page_get_page_no(page) == page_id.page_no());
     max_lsn= lsn;
     result= page;
   }
 
   return result;
 }
+
+bool recv_dblwr_t::restore_first_page(ulint space_id, const char *name,
+                                      os_file_t file)
+{
+  const page_id_t page_id(space_id, 0);
+  const byte* page= find_page(page_id);
+  if (!page)
+  {
+    /* If the first page of the given user tablespace is not there
+    in the doublewrite buffer, then the recovery is going to fail
+    now. Hence this is treated as error. */
+    ib::error()
+            << "Corrupted page " << page_id << " of datafile '"
+            << name <<"' could not be found in the doublewrite buffer.";
+    return true;
+  }
+
+  ulint physical_size= fil_space_t::physical_size(
+    mach_read_from_4(page + FSP_HEADER_OFFSET + FSP_SPACE_FLAGS));
+  ib::info() << "Restoring page " << page_id << " of datafile '"
+          << name << "' from the doublewrite buffer. Writing "
+          << physical_size << " bytes into file '" << name << "'";
+
+  return os_file_write(
+           IORequestWrite, name, file, page, 0, physical_size) !=
+         DB_SUCCESS;
+}

storage/innobase/srv/srv0start.cc

@@ -484,7 +484,8 @@ static ulint trx_rseg_get_n_undo_tablespaces()
 @param[in]	name	tablespace file name
 @param[in]	i	undo tablespace count
 @return undo tablespace identifier
-@retval 0 on failure */
+@retval 0               if file doesn't exist
+@retval ULINT_UNDEFINED if page0 is corrupted */
 static ulint srv_undo_tablespace_open(bool create, const char* name, ulint i)
 {
   bool success;
@@ -523,13 +524,13 @@ static ulint srv_undo_tablespace_open(bool create, const char* name, ulint i)
   {
     page_t *page= static_cast<byte*>(aligned_malloc(srv_page_size,
                                                     srv_page_size));
-    dberr_t err= os_file_read(IORequestRead, fh, page, 0, srv_page_size);
-    if (err != DB_SUCCESS)
+    if (os_file_read(IORequestRead, fh, page, 0, srv_page_size) !=
+        DB_SUCCESS)
     {
 err_exit:
       ib::error() << "Unable to read first page of file " << name;
       aligned_free(page);
-      return err;
+      return ULINT_UNDEFINED;
     }
 
     uint32_t id= mach_read_from_4(FIL_PAGE_SPACE_ID + page);
@@ -538,19 +539,18 @@ static ulint srv_undo_tablespace_open(bool create, const char* name, ulint i)
                           FSP_HEADER_OFFSET + FSP_SPACE_ID + page, 4))
     {
       ib::error() << "Inconsistent tablespace ID in file " << name;
-      err= DB_CORRUPTION;
       goto err_exit;
     }
 
+    space_id= id;
     fsp_flags= mach_read_from_4(FSP_HEADER_OFFSET + FSP_SPACE_FLAGS + page);
     if (buf_page_is_corrupted(false, page, fsp_flags))
     {
       ib::error() << "Checksum mismatch in the first page of file " << name;
-      err= DB_CORRUPTION;
-      goto err_exit;
+      if (recv_sys.dblwr.restore_first_page(space_id, name, fh))
+        goto err_exit;
     }
 
-    space_id= id;
     snprintf(undo_name, sizeof undo_name, "innodb_undo%03u", id);
     aligned_free(page);
   }
@@ -669,17 +669,19 @@ static dberr_t srv_all_undo_tablespaces_open(bool create_new_db, ulint n_undo)
     snprintf(name, sizeof name, "%s%cundo%03zu", srv_undo_dir,
              OS_PATH_SEPARATOR, i + 1);
     ulint space_id= srv_undo_tablespace_open(create_new_db, name, i);
-    if (!space_id)
-    {
+    switch (space_id) {
+    case ULINT_UNDEFINED:
+      return DB_CORRUPTION;
+    case 0:
       if (!create_new_db)
-        break;
-      ib::error() << "Unable to open create tablespace '" << name << "'.";
+        goto unused_undo;
+      sql_print_error("InnoDB: Unable to open create tablespace '%s'", name);
       return DB_ERROR;
+    default:
+      /* Should be no gaps in undo tablespace ids. */
+      ut_a(!i || prev_id + 1 == space_id);
     }
 
-    /* Should be no gaps in undo tablespace ids. */
-    ut_a(!i || prev_id + 1 == space_id);
-
     prev_id= space_id;
 
     /* Note the first undo tablespace id in case of
@@ -692,14 +694,15 @@ static dberr_t srv_all_undo_tablespaces_open(bool create_new_db, ulint n_undo)
   We stop at the first failure. These are undo tablespaces that are
   not in use and therefore not required by recovery. We only check
   that there are no gaps. */
-
+unused_undo:
   for (ulint i= prev_id + 1; i < srv_undo_space_id_start + TRX_SYS_N_RSEGS;
        ++i)
   {
      char name[OS_FILE_MAX_PATH];
      snprintf(name, sizeof(name),
               "%s%cundo%03zu", srv_undo_dir, OS_PATH_SEPARATOR, i);
-     if (!srv_undo_tablespace_open(create_new_db, name, i))
+     ulint space_id= srv_undo_tablespace_open(create_new_db, name, i);
+     if (!space_id || space_id == SRV_SPACE_ID_UPPER_BOUND)
        break;
      ++srv_undo_tablespaces_open;
   }