1 files changed, 187 insertions, 224 deletions

diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 60ede653bee..736aa32a4c2 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -67,6 +67,7 @@ Created 11/5/1995 Heikki Tuuri
 #include "buf0dump.h"
 #include <map>
 #include <sstream>
+#include "log.h"
 
 using st_::span;
 
@@ -583,9 +584,6 @@ bool buf_is_zeroes(span<const byte> buf)
 bool buf_page_is_corrupted(bool check_lsn, const byte *read_buf,
                            uint32_t fsp_flags)
 {
-#ifndef UNIV_INNOCHECKSUM
-	DBUG_EXECUTE_IF("buf_page_import_corrupt_failure", return(true); );
-#endif
 	if (fil_space_t::full_crc32(fsp_flags)) {
 		bool compressed = false, corrupted = false;
 		const uint size = buf_page_full_crc32_size(
@@ -2197,6 +2195,7 @@ void buf_page_free(fil_space_t *space, uint32_t page, mtr_t *mtr)
   ++buf_pool.stat.n_page_gets;
   const page_id_t page_id(space->id, page);
   buf_pool_t::hash_chain &chain= buf_pool.page_hash.cell_get(page_id.fold());
+  uint32_t fix;
   buf_block_t *block;
   {
     transactional_shared_lock_guard<page_hash_latch> g
@@ -2209,7 +2208,13 @@ void buf_page_free(fil_space_t *space, uint32_t page, mtr_t *mtr)
     /* To avoid a deadlock with buf_LRU_free_page() of some other page
     and buf_page_write_complete() of this page, we must not wait for a
     page latch while holding a page_hash latch. */
-    block->page.fix();
+    fix= block->page.fix();
+  }
+
+  if (UNIV_UNLIKELY(fix < buf_page_t::UNFIXED))
+  {
+    block->page.unfix();
+    return;
   }
 
   block->page.lock.x_lock();
@@ -2402,6 +2407,7 @@ buf_zip_decompress(
 	case FIL_PAGE_RTREE:
 		if (page_zip_decompress(&block->page.zip,
 					block->page.frame, TRUE)) {
+func_exit:
 			if (space) {
 				space->release();
 			}
@@ -2421,11 +2427,7 @@ buf_zip_decompress(
 	case FIL_PAGE_TYPE_ZBLOB2:
 		/* Copy to uncompressed storage. */
 		memcpy(block->page.frame, frame, block->zip_size());
-		if (space) {
-			space->release();
-		}
-
-		return(TRUE);
+		goto func_exit;
 	}
 
 	ib::error() << "Unknown compressed page type "
@@ -2440,12 +2442,6 @@ err_exit:
 	}
 
 	if (space) {
-		if (encrypted) {
-			dict_set_encrypted_by_space(space);
-		} else {
-			dict_set_corrupted_by_space(space);
-		}
-
 		space->release();
 	}
 
@@ -2458,7 +2454,7 @@ err_exit:
 @param[in]	rw_latch		RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH
 @param[in]	guess			guessed block or NULL
 @param[in]	mode			BUF_GET, BUF_GET_IF_IN_POOL,
-BUF_PEEK_IF_IN_POOL, BUF_GET_NO_LATCH, or BUF_GET_IF_IN_POOL_OR_WATCH
+BUF_PEEK_IF_IN_POOL, or BUF_GET_IF_IN_POOL_OR_WATCH
 @param[in]	mtr			mini-transaction
 @param[out]	err			DB_SUCCESS or error code
 @param[in]	allow_ibuf_merge	Allow change buffer merge to happen
@@ -2481,8 +2477,8 @@ buf_page_get_low(
 	unsigned	access_time;
 	ulint		retries = 0;
 
-	ut_ad((mtr == NULL) == (mode == BUF_EVICT_IF_IN_POOL));
 	ut_ad(!mtr || mtr->is_active());
+	ut_ad(mtr || mode == BUF_PEEK_IF_IN_POOL);
 	ut_ad((rw_latch == RW_S_LATCH)
 	      || (rw_latch == RW_X_LATCH)
 	      || (rw_latch == RW_SX_LATCH)
@@ -2499,24 +2495,15 @@ buf_page_get_low(
 
 #ifdef UNIV_DEBUG
 	switch (mode) {
-	case BUF_EVICT_IF_IN_POOL:
-		/* After DISCARD TABLESPACE, the tablespace would not exist,
-		but in IMPORT TABLESPACE, PageConverter::operator() must
-		replace any old pages, which were not evicted during DISCARD.
-		Skip the assertion on space_page_size. */
-		break;
 	case BUF_PEEK_IF_IN_POOL:
 	case BUF_GET_IF_IN_POOL:
 		/* The caller may pass a dummy page size,
 		because it does not really matter. */
 		break;
 	default:
-		ut_error;
+		MY_ASSERT_UNREACHABLE();
 	case BUF_GET_POSSIBLY_FREED:
 		break;
-	case BUF_GET_NO_LATCH:
-		ut_ad(rw_latch == RW_NO_LATCH);
-		/* fall through */
 	case BUF_GET:
 	case BUF_GET_IF_IN_POOL_OR_WATCH:
 		ut_ad(!mtr->is_freeing_tree());
@@ -2573,7 +2560,6 @@ loop:
 	switch (mode) {
 	case BUF_GET_IF_IN_POOL:
 	case BUF_PEEK_IF_IN_POOL:
-	case BUF_EVICT_IF_IN_POOL:
 		return nullptr;
 	case BUF_GET_IF_IN_POOL_OR_WATCH:
 		/* We cannot easily use a memory transaction here. */
@@ -2603,62 +2589,15 @@ loop:
 	checksum cannot be decypted. */
 
 	if (dberr_t local_err = buf_read_page(page_id, zip_size)) {
-		if (mode == BUF_GET_POSSIBLY_FREED) {
-			if (err) {
-				*err = local_err;
-			}
-			return nullptr;
-		} else if (retries < BUF_PAGE_READ_MAX_RETRIES) {
-			++retries;
-			DBUG_EXECUTE_IF("innodb_page_corruption_retries",
+		if (mode != BUF_GET_POSSIBLY_FREED
+		    && retries++ < BUF_PAGE_READ_MAX_RETRIES) {
+			DBUG_EXECUTE_IF("intermittent_read_failure",
 					retries = BUF_PAGE_READ_MAX_RETRIES;);
 		} else {
 			if (err) {
 				*err = local_err;
 			}
-			/* Pages whose encryption key is unavailable or the
-			configured key, encryption algorithm or encryption
-			method are incorrect are marked as encrypted in
-			buf_page_check_corrupt(). Unencrypted page could be
-			corrupted in a way where the key_id field is
-			nonzero. There is no checksum on field
-			FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION. */
-			switch (local_err) {
-			case DB_PAGE_CORRUPTED:
-				if (!srv_force_recovery) {
-					break;
-				}
-				/* fall through */
-			case DB_DECRYPTION_FAILED:
-				return nullptr;
-			default:
-				break;
-			}
-
-			/* Try to set table as corrupted instead of
-			asserting. */
-			if (page_id.space() == TRX_SYS_SPACE) {
-			} else if (page_id.space() == SRV_TMP_SPACE_ID) {
-			} else if (fil_space_t* space
-				   = fil_space_t::get(page_id.space())) {
-				bool set = dict_set_corrupted_by_space(space);
-				space->release();
-				if (set) {
-					return nullptr;
-				}
-			}
-
-			if (local_err == DB_IO_ERROR) {
-				return nullptr;
-			}
-
-			ib::fatal() << "Unable to read page " << page_id
-				    << " into the buffer pool after "
-				    << BUF_PAGE_READ_MAX_RETRIES
-				    << ". The most probable cause"
-				" of this error may be that the"
-				" table has been corrupted."
-				" See https://mariadb.com/kb/en/library/innodb-recovery-modes/";
+			return nullptr;
 		}
 	} else {
 		buf_read_ahead_random(page_id, zip_size, ibuf_inside(mtr));
@@ -2673,10 +2612,14 @@ got_block:
 	ut_ad(state > buf_page_t::FREED);
 
 	if (state > buf_page_t::READ_FIX && state < buf_page_t::WRITE_FIX) {
-		if (mode == BUF_PEEK_IF_IN_POOL
-		    || mode == BUF_EVICT_IF_IN_POOL) {
+		if (mode == BUF_PEEK_IF_IN_POOL) {
 ignore_block:
+			ut_ad(mode == BUF_GET_POSSIBLY_FREED
+			      || mode == BUF_PEEK_IF_IN_POOL);
 			block->unfix();
+			if (err) {
+				*err = DB_CORRUPTION;
+			}
 			return nullptr;
 		}
 
@@ -2702,46 +2645,29 @@ ignore_block:
 				*err = DB_PAGE_CORRUPTED;
 			}
 
-			if (page_id.space() == TRX_SYS_SPACE) {
-			} else if (page_id.space() == SRV_TMP_SPACE_ID) {
-			} else if (fil_space_t* space =
-				   fil_space_t::get(page_id.space())) {
-				bool set = dict_set_corrupted_by_space(space);
-				space->release();
-				if (set) {
-					return nullptr;
-				}
-			}
-
-			ib::fatal() << "Unable to read page " << page_id
-				    << " into the buffer pool after "
-				    << BUF_PAGE_READ_MAX_RETRIES
-				    << ". The most probable cause"
-				" of this error may be that the"
-				" table has been corrupted."
-				" See https://mariadb.com/kb/en/library/innodb-recovery-modes/";
-		}
-	} else if (mode == BUF_PEEK_IF_IN_POOL) {
-		if (UNIV_UNLIKELY(!block->page.frame)) {
-			/* This mode is only used for dropping an
-			adaptive hash index. There cannot be an
-			adaptive hash index for a compressed-only page. */
-			goto ignore_block;
+			return nullptr;
 		}
-	} else if (mode == BUF_EVICT_IF_IN_POOL) {
+	} else if (mode != BUF_PEEK_IF_IN_POOL) {
+        } else if (!mtr) {
 		ut_ad(!block->page.oldest_modification());
 		mysql_mutex_lock(&buf_pool.mutex);
 		block->unfix();
 
-		if (!buf_LRU_free_page(&block->page, true)) {
+			if (!buf_LRU_free_page(&block->page, true)) {
 			ut_ad(0);
 		}
 
 		mysql_mutex_unlock(&buf_pool.mutex);
 		return nullptr;
+	} else if (UNIV_UNLIKELY(!block->page.frame)) {
+		/* The BUF_PEEK_IF_IN_POOL mode is mainly used for dropping an
+		adaptive hash index. There cannot be an
+		adaptive hash index for a compressed-only page. */
+		goto ignore_block;
 	}
 
-	ut_ad(mode == BUF_GET_IF_IN_POOL || block->zip_size() == zip_size);
+	ut_ad(mode == BUF_GET_IF_IN_POOL || mode == BUF_PEEK_IF_IN_POOL
+	      || block->zip_size() == zip_size);
 
 	if (UNIV_UNLIKELY(!block->page.frame)) {
 		if (!block->page.lock.x_lock_try()) {
@@ -2786,14 +2712,11 @@ wait_for_unfix:
 
 			if (state < buf_page_t::UNFIXED + 1) {
 				ut_ad(state > buf_page_t::FREED);
-				ut_ad(mode == BUF_GET_POSSIBLY_FREED
-				      || mode == BUF_PEEK_IF_IN_POOL);
-				block->page.unfix();
 				block->page.lock.x_unlock();
 				hash_lock.unlock();
 				buf_LRU_block_free_non_file_page(new_block);
 				mysql_mutex_unlock(&buf_pool.mutex);
-				return nullptr;
+				goto ignore_block;
 			}
 
 			mysql_mutex_unlock(&buf_pool.mutex);
@@ -2911,60 +2834,135 @@ re_evict:
 #endif /* UNIV_DEBUG || UNIV_IBUF_DEBUG */
 
 	ut_ad(state > buf_page_t::FREED);
-	ut_ad(state < buf_page_t::UNFIXED || (~buf_page_t::LRU_MASK) & state);
-	ut_ad(state > buf_page_t::WRITE_FIX || state < buf_page_t::READ_FIX);
-
-	/* While tablespace is reinited the indexes are already freed but the
-	blocks related to it still resides in buffer pool. Trying to remove
-	such blocks from buffer pool would invoke removal of AHI entries
-	associated with these blocks. Logic to remove AHI entry will try to
-	load the block but block is already in free state. Handle the said case
-	with mode = BUF_PEEK_IF_IN_POOL that is invoked from
-	"btr_search_drop_page_hash_when_freed". */
-	ut_ad(mode == BUF_GET_POSSIBLY_FREED || mode == BUF_PEEK_IF_IN_POOL
-	      || state > buf_page_t::UNFIXED);
-
-	const bool not_first_access = block->page.set_accessed();
-
-	if (mode != BUF_PEEK_IF_IN_POOL) {
-		buf_page_make_young_if_needed(&block->page);
-		if (!not_first_access) {
-			buf_read_ahead_linear(page_id, block->zip_size(),
-					      ibuf_inside(mtr));
-		}
+	if (UNIV_UNLIKELY(state < buf_page_t::UNFIXED)) {
+		goto ignore_block;
 	}
+	ut_ad((~buf_page_t::LRU_MASK) & state);
+	ut_ad(state > buf_page_t::WRITE_FIX || state < buf_page_t::READ_FIX);
 
 #ifdef UNIV_DEBUG
 	if (!(++buf_dbg_counter % 5771)) buf_pool.validate();
 #endif /* UNIV_DEBUG */
 	ut_ad(block->page.frame);
 
-	ut_ad(block->page.id() == page_id);
-
 	if (state >= buf_page_t::UNFIXED
 	    && allow_ibuf_merge
 	    && fil_page_get_type(block->page.frame) == FIL_PAGE_INDEX
 	    && page_is_leaf(block->page.frame)) {
 		block->page.lock.x_lock();
+		ut_ad(block->page.id() == page_id
+		      || (state >= buf_page_t::READ_FIX
+			  && state < buf_page_t::WRITE_FIX));
+
+#ifdef BTR_CUR_HASH_ADAPT
+		if (dict_index_t* index = block->index) {
+			if (index->freed()) {
+				btr_search_drop_page_hash_index(block);
+			}
+		}
+#endif /* BTR_CUR_HASH_ADAPT */
+
+		dberr_t e;
+
+		if (UNIV_UNLIKELY(block->page.id() != page_id)) {
+page_id_mismatch:
+			state = block->page.state();
+			e = DB_CORRUPTION;
+ibuf_merge_corrupted:
+			if (err) {
+				*err = e;
+			}
+
+			buf_pool.corrupted_evict(&block->page, state);
+			return nullptr;
+		}
+
 		state = block->page.state();
 		ut_ad(state < buf_page_t::READ_FIX);
 
 		if (state >= buf_page_t::IBUF_EXIST
 		    && state < buf_page_t::REINIT) {
 			block->page.clear_ibuf_exist();
-			ibuf_merge_or_delete_for_page(block, page_id,
-						      block->zip_size());
+			e = ibuf_merge_or_delete_for_page(block, page_id,
+							  block->zip_size());
+			if (UNIV_UNLIKELY(e != DB_SUCCESS)) {
+				goto ibuf_merge_corrupted;
+			}
 		}
 
 		if (rw_latch == RW_X_LATCH) {
 			mtr->memo_push(block, MTR_MEMO_PAGE_X_FIX);
+			goto got_latch;
 		} else {
 			block->page.lock.x_unlock();
 			goto get_latch;
 		}
 	} else {
 get_latch:
-		mtr->page_lock(block, rw_latch);
+		switch (rw_latch) {
+			mtr_memo_type_t fix_type;
+		case RW_NO_LATCH:
+			mtr->memo_push(block, MTR_MEMO_BUF_FIX);
+			return block;
+		case RW_S_LATCH:
+			fix_type = MTR_MEMO_PAGE_S_FIX;
+			block->page.lock.s_lock();
+			ut_ad(!block->page.is_read_fixed());
+			if (UNIV_UNLIKELY(block->page.id() != page_id)) {
+				block->page.lock.s_unlock();
+				block->page.lock.x_lock();
+				goto page_id_mismatch;
+			}
+get_latch_valid:
+#ifdef BTR_CUR_HASH_ADAPT
+			if (dict_index_t* index = block->index) {
+				if (index->freed()) {
+					mtr_t::defer_drop_ahi(block, fix_type);
+				}
+			}
+#endif /* BTR_CUR_HASH_ADAPT */
+			mtr->memo_push(block, fix_type);
+			break;
+		case RW_SX_LATCH:
+			fix_type = MTR_MEMO_PAGE_SX_FIX;
+			block->page.lock.u_lock();
+			ut_ad(!block->page.is_io_fixed());
+			if (UNIV_UNLIKELY(block->page.id() != page_id)) {
+				block->page.lock.u_x_upgrade();
+				goto page_id_mismatch;
+			}
+			goto get_latch_valid;
+		default:
+			ut_ad(rw_latch == RW_X_LATCH);
+			fix_type = MTR_MEMO_PAGE_X_FIX;
+			if (block->page.lock.x_lock_upgraded()) {
+				ut_ad(block->page.id() == page_id);
+				block->unfix();
+				mtr->page_lock_upgrade(*block);
+				return block;
+			}
+			if (UNIV_UNLIKELY(block->page.id() != page_id)) {
+				goto page_id_mismatch;
+			}
+			goto get_latch_valid;
+		}
+
+got_latch:
+		ut_ad(page_id_t(page_get_space_id(block->page.frame),
+				page_get_page_no(block->page.frame))
+		      == page_id);
+
+		if (mode == BUF_GET_POSSIBLY_FREED
+		    || mode == BUF_PEEK_IF_IN_POOL) {
+			return block;
+		}
+
+		const bool not_first_access{block->page.set_accessed()};
+		buf_page_make_young_if_needed(&block->page);
+		if (!not_first_access) {
+			buf_read_ahead_linear(page_id, block->zip_size(),
+					      ibuf_inside(mtr));
+		}
 	}
 
 	return block;
@@ -2976,8 +2974,8 @@ get_latch:
 @param[in]	rw_latch		RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH
 @param[in]	guess			guessed block or NULL
 @param[in]	mode			BUF_GET, BUF_GET_IF_IN_POOL,
-BUF_PEEK_IF_IN_POOL, BUF_GET_NO_LATCH, or BUF_GET_IF_IN_POOL_OR_WATCH
-@param[in]	mtr			mini-transaction
+BUF_PEEK_IF_IN_POOL, or BUF_GET_IF_IN_POOL_OR_WATCH
+@param[in,out]	mtr			mini-transaction, or NULL
 @param[out]	err			DB_SUCCESS or error code
 @param[in]	allow_ibuf_merge	Allow change buffer merge while
 reading the pages from file.
@@ -2995,6 +2993,13 @@ buf_page_get_gen(
 {
   if (buf_block_t *block= recv_sys.recover(page_id))
   {
+    if (UNIV_UNLIKELY(block == reinterpret_cast<buf_block_t*>(-1)))
+    {
+    corrupted:
+      if (err)
+        *err= DB_CORRUPTION;
+      return nullptr;
+    }
     /* Recovery is a special case; we fix() before acquiring lock. */
     auto s= block->page.fix();
     ut_ad(s >= buf_page_t::FREED);
@@ -3006,7 +3011,12 @@ buf_page_get_gen(
     const bool must_merge= allow_ibuf_merge &&
       ibuf_page_exists(page_id, block->zip_size());
     if (s < buf_page_t::UNFIXED)
+    {
+    got_freed_page:
       ut_ad(mode == BUF_GET_POSSIBLY_FREED || mode == BUF_PEEK_IF_IN_POOL);
+      block->page.unfix();
+      goto corrupted;
+    }
     else if (must_merge &&
              fil_page_get_type(block->page.frame) == FIL_PAGE_INDEX &&
              page_is_leaf(block->page.frame))
@@ -3016,12 +3026,22 @@ buf_page_get_gen(
       ut_ad(s > buf_page_t::FREED);
       ut_ad(s < buf_page_t::READ_FIX);
       if (s < buf_page_t::UNFIXED)
-        ut_ad(mode == BUF_GET_POSSIBLY_FREED || mode == BUF_PEEK_IF_IN_POOL);
+      {
+        block->page.lock.x_unlock();
+        goto got_freed_page;
+      }
       else
       {
         if (block->page.is_ibuf_exist())
           block->page.clear_ibuf_exist();
-        ibuf_merge_or_delete_for_page(block, page_id, block->zip_size());
+        if (dberr_t e=
+            ibuf_merge_or_delete_for_page(block, page_id, block->zip_size()))
+        {
+          if (err)
+            *err= e;
+          buf_pool.corrupted_evict(&block->page, s);
+          return nullptr;
+        }
       }
 
       if (rw_latch == RW_X_LATCH)
@@ -3156,7 +3176,7 @@ buf_block_t *buf_page_try_get(const page_id_t page_id, mtr_t *mtr)
 
   block->page.fix();
   ut_ad(!block->page.is_read_fixed());
-  mtr_memo_push(mtr, block, MTR_MEMO_PAGE_S_FIX);
+  mtr->memo_push(block, MTR_MEMO_PAGE_S_FIX);
 
 #ifdef UNIV_DEBUG
   if (!(++buf_dbg_counter % 5771)) buf_pool.validate();
@@ -3228,7 +3248,7 @@ static buf_block_t *buf_page_create_low(page_id_t page_id, ulint zip_size,
       {
         mysql_mutex_unlock(&buf_pool.mutex);
         buf_block_t *block= reinterpret_cast<buf_block_t*>(bpage);
-        mtr_memo_push(mtr, block, MTR_MEMO_PAGE_X_FIX);
+        mtr->memo_push(block, MTR_MEMO_PAGE_X_FIX);
 #ifdef BTR_CUR_HASH_ADAPT
         drop_hash_entry= block->index;
 #endif
@@ -3268,7 +3288,7 @@ static buf_block_t *buf_page_create_low(page_id_t page_id, ulint zip_size,
         bpage->lock.free();
 #endif
         ut_free(bpage);
-        mtr_memo_push(mtr, free_block, MTR_MEMO_PAGE_X_FIX);
+        mtr->memo_push(free_block, MTR_MEMO_PAGE_X_FIX);
         bpage= &free_block->page;
       }
     }
@@ -3483,37 +3503,6 @@ ATTRIBUTE_COLD void buf_page_monitor(const buf_page_t &bpage, bool read)
 	MONITOR_INC_NOCHECK(counter);
 }
 
-/** Mark a table corrupted.
-@param[in]	bpage	corrupted page
-@param[in]	space	tablespace of the corrupted page */
-ATTRIBUTE_COLD
-static void buf_mark_space_corrupt(buf_page_t* bpage, const fil_space_t& space)
-{
-	/* 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 (!space.crypt_data
-	    || space.crypt_data->type == CRYPT_SCHEME_UNENCRYPTED) {
-		dict_set_corrupted_by_space(&space);
-	} else {
-		dict_set_encrypted_by_space(&space);
-	}
-}
-
-/** Mark a table corrupted.
-@param[in]	bpage	Corrupted page
-@param[in]	node	data file
-Also remove the bpage from LRU list. */
-ATTRIBUTE_COLD
-static void buf_corrupt_page_release(buf_page_t *bpage, const fil_node_t &node)
-{
-  ut_ad(bpage->id().space() == node.space->id);
-  buf_pool.corrupted_evict(bpage);
-
-  if (!srv_force_recovery)
-    buf_mark_space_corrupt(bpage, *node.space);
-}
-
 /** Check if the encrypted page is corrupted for the full crc32 format.
 @param[in]	space_id	page belongs to space id
 @param[in]	d		page
@@ -3589,21 +3578,8 @@ static dberr_t buf_page_check_corrupt(buf_page_t *bpage,
 		ib::error()
 			<< "The page " << bpage->id()
 			<< " in file '" << node.name
-			<< "' cannot be decrypted.";
-
-		ib::info()
-			<< "However key management plugin or used key_version "
-			<< key_version
-			<< " is not found or"
-			" used encryption algorithm or method does not match.";
-
-		if (bpage->id().space() != TRX_SYS_SPACE) {
-			ib::info()
-				<< "Marking tablespace as missing."
-				" You may drop this table or"
-				" install correct key management plugin"
-				" and key file.";
-		}
+			<< "' cannot be decrypted; key_version="
+			<< key_version;
 	}
 
 	return (err);
@@ -3623,6 +3599,10 @@ dberr_t buf_page_t::read_complete(const fil_node_t &node)
   ut_ad(zip_size() == node.space->zip_size());
   ut_ad(!!zip.ssize == !!zip.data);
 
+  ut_d(auto n=) buf_pool.n_pend_reads--;
+  ut_ad(n > 0);
+  buf_pool.stat.n_pages_read++;
+
   const byte *read_frame= zip.data ? zip.data : frame;
   ut_ad(read_frame);
 
@@ -3653,7 +3633,11 @@ dberr_t buf_page_t::read_complete(const fil_node_t &node)
 
     if (read_id == expected_id);
     else if (read_id == page_id_t(0, 0))
-      /* This is likely an uninitialized page. */;
+    {
+      /* This is likely an uninitialized (all-zero) page. */
+      err= DB_FAIL;
+      goto release_page;
+    }
     else if (!node.space->full_crc32() &&
              page_id_t(0, read_id.page_no()) == expected_id)
       /* FIL_PAGE_SPACE_ID was written as garbage in the system tablespace
@@ -3669,25 +3653,18 @@ dberr_t buf_page_t::read_complete(const fil_node_t &node)
       goto release_page;
     }
     else
+    {
       ib::error() << "Space id and page no stored in the page, read in are "
                   << read_id << ", should be " << expected_id;
+      err= DB_PAGE_CORRUPTED;
+      goto release_page;
+    }
   }
 
   err= buf_page_check_corrupt(this, node);
   if (UNIV_UNLIKELY(err != DB_SUCCESS))
   {
 database_corrupted:
-    /* Not a real corruption if it was triggered by error injection */
-    DBUG_EXECUTE_IF("buf_page_import_corrupt_failure",
-                    if (!is_predefined_tablespace(id().space()))
-                    {
-                      buf_corrupt_page_release(this, node);
-                      ib::info() << "Simulated IMPORT corruption";
-                      return err;
-                    }
-                    err= DB_SUCCESS;
-                    goto page_not_corrupt;);
-
     if (belongs_to_unzip_LRU())
       memset_aligned<UNIV_PAGE_SIZE_MIN>(frame, 0, srv_page_size);
 
@@ -3700,38 +3677,28 @@ database_corrupted:
 
       buf_page_print(read_frame, zip_size());
 
+      node.space->set_corrupted();
+
       ib::info() << " You can use CHECK TABLE to scan"
                     " your table for corruption. "
                  << FORCE_RECOVERY_MSG;
     }
 
     if (!srv_force_recovery)
-    {
-      /* If the corruption is in the system tablespace, we will
-      intentionally crash the server. */
-      if (expected_id.space() == TRX_SYS_SPACE)
-        ib::fatal() << "Aborting because of a corrupt database page.";
-      buf_corrupt_page_release(this, node);
-      return err;
-    }
+      goto release_page;
   }
 
-  DBUG_EXECUTE_IF("buf_page_import_corrupt_failure",
-                  page_not_corrupt: err= err; );
-
   if (err == DB_PAGE_CORRUPTED || err == DB_DECRYPTION_FAILED)
   {
 release_page:
-    buf_corrupt_page_release(this, node);
-    if (recv_recovery_is_on())
-      recv_sys.free_corrupted_page(expected_id);
+    buf_pool.corrupted_evict(this, buf_page_t::READ_FIX);
     return err;
   }
 
   const bool recovery= recv_recovery_is_on();
 
-  if (recovery)
-    recv_recover_page(node.space, this);
+  if (recovery && !recv_recover_page(node.space, this))
+    return DB_PAGE_CORRUPTED;
 
   const bool ibuf_may_exist= frame && !recv_no_ibuf_operations &&
     (!expected_id.space() || !is_predefined_tablespace(expected_id.space())) &&
@@ -3755,10 +3722,6 @@ release_page:
 
   lock.x_unlock(true);
 
-  ut_d(auto n=) buf_pool.n_pend_reads--;
-  ut_ad(n > 0);
-  buf_pool.stat.n_pages_read++;
-
   return DB_SUCCESS;
 }