MDEV-19541 InnoDB crashes when trying to recover a corrupted page

- Don't apply redo log for the corrupted page when innodb_force_recovery > 0.
- Allow the table to be dropped when index root page is
corrupted when innodb_force_recovery > 0.

mysql-test/suite/encryption/r/corrupted_during_recovery.result

@@ -9,14 +9,11 @@ INSERT INTO t2 VALUES(2);
 SELECT * FROM t1;
 ERROR 42000: Unknown storage engine 'InnoDB'
 SELECT * FROM t1;
-a
-1
-2
+ERROR 42S02: Table 'test.t1' doesn't exist in engine
 SELECT * FROM t2;
 a
 2
-CHECK TABLE t1,t2;
+CHECK TABLE t2;
 Table	Op	Msg_type	Msg_text
-test.t1	check	status	OK
 test.t2	check	status	OK
 DROP TABLE t1, t2;

mysql-test/suite/encryption/t/corrupted_during_recovery.test

@@ -50,11 +50,16 @@ EOF
 --source include/start_mysqld.inc
 --error ER_UNKNOWN_STORAGE_ENGINE
 SELECT * FROM t1;
+--disable_query_log
+call mtr.add_suppression("InnoDB: Encrypted page \\[page id: space=[1-9][0-9]*, page number=3\\] in file .*test.t[1].ibd looks corrupted; key_version=1786080875");
+call mtr.add_suppression("InnoDB: Table `test`\\.`t1` is corrupted. Please drop the table and recreate.");
+--enable_query_log
 let $restart_parameters=--innodb_force_recovery=1;
 --source include/restart_mysqld.inc
 
+--error ER_NO_SUCH_TABLE_IN_ENGINE
 SELECT * FROM t1;
 SELECT * FROM t2;
-CHECK TABLE t1,t2;
+CHECK TABLE t2;
 
 DROP TABLE t1, t2;

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

@@ -9,14 +9,11 @@ INSERT INTO t2 VALUES(1);
 SELECT * FROM t1;
 ERROR 42000: Unknown storage engine 'InnoDB'
 SELECT * FROM t1;
-a
-0
-2
+ERROR 42S02: Table 'test.t1' doesn't exist in engine
 SELECT * FROM t2;
 a
 1
-CHECK TABLE t1,t2;
+CHECK TABLE t2;
 Table	Op	Msg_type	Msg_text
-test.t1	check	status	OK
 test.t2	check	status	OK
 DROP TABLE t1, t2;

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

@@ -6,6 +6,8 @@ call mtr.add_suppression("Plugin 'InnoDB' init function returned error");
 call mtr.add_suppression("Plugin 'InnoDB' registration as a STORAGE ENGINE failed");
 call mtr.add_suppression("InnoDB: Database page corruption on disk or a failed file read of tablespace test/t1 page");
 call mtr.add_suppression("InnoDB: Failed to read file '.*test.t1\\.ibd' at offset 3: Page read from tablespace is corrupted.");
+call mtr.add_suppression("InnoDB: Background Page read failed to read or decrypt \\[page id: space=\\d+, page number=3\\]");
+call mtr.add_suppression("InnoDB: Table `test`.`t1` is corrupted. Please drop the table and recreate.");
 --enable_query_log
 
 let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
@@ -53,8 +55,10 @@ EOF
 SELECT * FROM t1;
 let $restart_parameters=--innodb_force_recovery=1;
 --source include/restart_mysqld.inc
+
+--error ER_NO_SUCH_TABLE_IN_ENGINE
 SELECT * FROM t1;
 SELECT * FROM t2;
-CHECK TABLE t1,t2;
+CHECK TABLE t2;
 
 DROP TABLE t1, t2;

storage/innobase/buf/buf0buf.cc

@@ -4378,6 +4378,11 @@ buf_page_get_gen(
 				return (NULL);
 			}
 
+			if (local_err == DB_PAGE_CORRUPTED
+			    && srv_force_recovery) {
+				return NULL;
+			}
+
 			/* Try to set table as corrupted instead of
 			asserting. */
 			if (page_id.space() != TRX_SYS_SPACE &&
@@ -5743,18 +5748,33 @@ buf_page_monitor(
 	MONITOR_INC_NOCHECK(counter);
 }
 
-/********************************************************************//**
-Mark a table with the specified space pointed by bpage->id.space() corrupted.
-Also remove the bpage from LRU list.
-@param[in,out]		bpage			Block */
+/** Mark a table corrupted.
+@param[in]	bpage	Corrupted page. */
 static
 void
 buf_mark_space_corrupt(buf_page_t* bpage)
+{
+	/* If block is not encrypted find the table with specified
+	space id, and mark it corrupted. Encrypted tables
+	are marked unusable later e.g. in ::open(). */
+	if (!bpage->encrypted) {
+		dict_set_corrupted_by_space(bpage->id.space());
+	} else {
+		dict_set_encrypted_by_space(bpage->id.space());
+	}
+}
+
+/** Mark a table corrupted.
+@param[in]	bpage	Corrupted page
+@param[in]	space	Corrupted page belongs to tablespace
+Also remove the bpage from LRU list. */
+static
+void
+buf_corrupt_page_release(buf_page_t* bpage, const fil_space_t* space)
 {
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 	const ibool	uncompressed = (buf_page_get_state(bpage)
 					== BUF_BLOCK_FILE_PAGE);
-	uint32_t	space = bpage->id.space();
 
 	/* First unfix and release lock on the bpage */
 	buf_pool_mutex_enter(buf_pool);
@@ -5773,13 +5793,8 @@ buf_mark_space_corrupt(buf_page_t* bpage)
 
 	mutex_exit(buf_page_get_mutex(bpage));
 
-	/* If block is not encrypted find the table with specified
-	space id, and mark it corrupted. Encrypted tables
-	are marked unusable later e.g. in ::open(). */
-	if (!bpage->encrypted) {
-		dict_set_corrupted_by_space(space);
-	} else {
-		dict_set_encrypted_by_space(space);
+	if (!srv_force_recovery) {
+		buf_mark_space_corrupt(bpage);
 	}
 
 	/* After this point bpage can't be referenced. */
@@ -5981,7 +5996,7 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 				"buf_page_import_corrupt_failure",
 				if (!is_predefined_tablespace(
 					    bpage->id.space())) {
-					buf_mark_space_corrupt(bpage);
+					buf_corrupt_page_release(bpage, space);
 					ib::info() << "Simulated IMPORT "
 						"corruption";
 					fil_space_release_for_io(space);
@@ -6015,7 +6030,7 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 					<< FORCE_RECOVERY_MSG;
 			}
 
-			if (srv_force_recovery < SRV_FORCE_IGNORE_CORRUPT) {
+			if (!srv_force_recovery) {
 
 				/* If page space id is larger than TRX_SYS_SPACE
 				(0), we will attempt to mark the corresponding
@@ -6025,7 +6040,7 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 						" a corrupt database page.";
 				}
 
-				buf_mark_space_corrupt(bpage);
+				buf_corrupt_page_release(bpage, space);
 				fil_space_release_for_io(space);
 				return(err);
 			}
@@ -6034,6 +6049,18 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 		DBUG_EXECUTE_IF("buf_page_import_corrupt_failure",
 				page_not_corrupt: bpage = bpage; );
 
+		if (err == DB_PAGE_CORRUPTED
+		    || err == DB_DECRYPTION_FAILED) {
+			buf_corrupt_page_release(bpage, space);
+
+			if (recv_recovery_is_on()) {
+				recv_recover_corrupt_page(bpage);
+			}
+
+			fil_space_release_for_io(space);
+			return err;
+		}
+
 		if (recv_recovery_is_on()) {
 			recv_recover_page(bpage);
 		}

storage/innobase/include/log0recv.h

@@ -49,6 +49,11 @@ dberr_t
 recv_find_max_checkpoint(ulint* max_field)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
+/** Reduces recv_sys->n_addrs for the corrupted page.
+This function should called when srv_force_recovery > 0.
+@param[in]	bpage	buffer pool page */
+void recv_recover_corrupt_page(buf_page_t* bpage);
+
 /** Apply any buffered redo log to a page that was just read from a data file.
 @param[in,out]	bpage	buffer pool page */
 ATTRIBUTE_COLD void recv_recover_page(buf_page_t* bpage);

storage/innobase/log/log0recv.cc

@@ -2212,6 +2212,32 @@ static void recv_recover_page(buf_block_t* block, mtr_t& mtr,
 	}
 }
 
+/** Reduces recv_sys->n_addrs for the corrupted page.
+This function should called when srv_force_recovery > 0.
+@param[in]	bpage	buffer pool page */
+void recv_recover_corrupt_page(buf_page_t* bpage)
+{
+	ut_ad(srv_force_recovery);
+	mutex_enter(&recv_sys->mutex);
+
+	if (!recv_sys->apply_log_recs) {
+		mutex_exit(&recv_sys->mutex);
+		return;
+	}
+
+	recv_addr_t* recv_addr = recv_get_fil_addr_struct(
+			bpage->id.space(), bpage->id.page_no());
+
+	ut_ad(recv_addr->state != RECV_WILL_NOT_READ);
+
+	if (recv_addr->state != RECV_BEING_PROCESSED
+	    && recv_addr->state != RECV_PROCESSED) {
+		recv_sys->n_addrs--;
+	}
+
+	mutex_exit(&recv_sys->mutex);
+}
+
 /** Apply any buffered redo log to a page that was just read from a data file.
 @param[in,out]	bpage	buffer pool page */
 void recv_recover_page(buf_page_t* bpage)