MDEV-24589 fixup: Crash on purge of SYS_INDEXES recordbb-10.5-MDEV-24589

A crash occurred during DROP TABLE in stress.ddl_innodb when
fil_delete_tablespace() for the DROP TABLE was waiting in
fil_check_pending_operations() and a purge thread for handling
an earlier DROP INDEX was attempting to load the index root page
in btr_free_if_exists() and btr_free_root_check(). The
buf_page_get_gen() would write out several times
"trying to read...being-dropped tablespace"
before giving up and committing suicide.

btr_free_if_exists(): Acquire a tablespace reference for the whole
duration of the operation, just in case.

btr_free_root_check(), xdes_lst_get_descriptor(),
fseg_inode_try_get(), flst_remove(),
fseg_free_step(), fseg_free_step_not_header(): Use BUF_GET_POSSIBLY_FREED
and be prepared for a nullptr return value in case
fil_check_pending_operations() had invoked fil_space_t::set_stopping().

fseg_free_page(): Add some fault tolerance for corrupted pages.
This is additional cleanup; this function is only used for
freeing individual pages when the tablespace is guaranteed to exist.

4 files changed, 120 insertions, 90 deletions

diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index de87ad02e68..28208abccfe 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -979,10 +979,14 @@ btr_free_root_check(
 	ut_ad(page_id.space() != SRV_TMP_SPACE_ID);
 	ut_ad(index_id != BTR_FREED_INDEX_ID);
 
-	buf_block_t*	block = buf_page_get(
-		page_id, zip_size, RW_X_LATCH, mtr);
+	buf_block_t*	block = buf_page_get_gen(
+		page_id, zip_size, RW_X_LATCH, NULL, BUF_GET_POSSIBLY_FREED,
+		__FILE__, __LINE__, mtr);
 
-	if (block) {
+	if (!block) {
+	} else if (block->page.status == buf_page_t::FREED) {
+		block = NULL;
+	} else {
 		buf_block_dbg_add_level(block, SYNC_TREE_NODE);
 
 		if (fil_page_index_page_check(block->frame)
@@ -1205,23 +1209,20 @@ top_loop:
 @param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
 @param[in]	index_id	PAGE_INDEX_ID contents
 @param[in,out]	mtr		mini-transaction */
-void
-btr_free_if_exists(
-	const page_id_t		page_id,
-	ulint			zip_size,
-	index_id_t		index_id,
-	mtr_t*			mtr)
+void btr_free_if_exists(const page_id_t page_id, ulint zip_size,
+                        index_id_t index_id, mtr_t *mtr)
 {
-	buf_block_t* root = btr_free_root_check(
-		page_id, zip_size, index_id, mtr);
-
-	if (root == NULL) {
-		return;
-	}
-
-	btr_free_but_not_root(root, mtr->get_log_mode());
-	mtr->set_named_space_id(page_id.space());
-	btr_free_root(root, mtr);
+  if (fil_space_t *space= fil_space_t::get(page_id.space()))
+  {
+    if (buf_block_t *root= btr_free_root_check(page_id, zip_size, index_id,
+                                               mtr))
+    {
+      btr_free_but_not_root(root, mtr->get_log_mode());
+      mtr->set_named_space(space);
+      btr_free_root(root, mtr);
+    }
+    space->release();
+  }
 }
 
 /** Free an index tree in a temporary tablespace.
diff --git a/storage/innobase/fsp/fsp0fsp.cc b/storage/innobase/fsp/fsp0fsp.cc
index ceaab79b1c4..cb9e0be428b 100644
--- a/storage/innobase/fsp/fsp0fsp.cc
+++ b/storage/innobase/fsp/fsp0fsp.cc
@@ -440,6 +440,7 @@ extent descriptor resides is x-locked.
 @param[in]	space		tablespace
 @param[in]	lst_node	file address of the list node
 				contained in the descriptor
+@param[in]	mode		BUF_GET or BUF_GET_POSSIBLY_FREED
 @param[out]	block		extent descriptor block
 @param[in,out]	mtr		mini-transaction
 @return pointer to the extent descriptor */
@@ -449,13 +450,14 @@ xdes_t*
 xdes_lst_get_descriptor(
 	const fil_space_t*	space,
 	fil_addr_t		lst_node,
+	ulint			mode,
 	buf_block_t**		block,
 	mtr_t*			mtr)
 {
-	ut_ad(mtr->memo_contains(*space));
-	return fut_get_ptr(space->id, space->zip_size(),
-			   lst_node, RW_SX_LATCH, mtr, block)
-		- XDES_FLST_NODE;
+  ut_ad(mtr->memo_contains(*space));
+  auto b= fut_get_ptr(space->id, space->zip_size(), lst_node, RW_SX_LATCH,
+                      mode, mtr, block);
+  return b ? b - XDES_FLST_NODE : nullptr;
 }
 
 /********************************************************************//**
@@ -995,8 +997,8 @@ fsp_alloc_free_extent(
 			}
 		}
 
-		descr = xdes_lst_get_descriptor(space, first, &desc_block,
-						mtr);
+		descr = xdes_lst_get_descriptor(space, first, BUF_GET,
+						&desc_block, mtr);
 	}
 
 	flst_remove(header, FSP_HEADER_OFFSET + FSP_FREE, desc_block,
@@ -1118,8 +1120,8 @@ fsp_alloc_free_page(
 					      descr - xdes->frame
 					      + XDES_FLST_NODE), mtr);
 		} else {
-			descr = xdes_lst_get_descriptor(space, first, &xdes,
-							mtr);
+			descr = xdes_lst_get_descriptor(space, first,
+							BUF_GET, &xdes, mtr);
 		}
 
 		/* Reset the hint */
@@ -1481,6 +1483,7 @@ static void fsp_free_seg_inode(
 @param[in]	header		segment header
 @param[in]	space		space id
 @param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
+@param[in]	mode		BUF_GET or BUF_GET_POSSIBLY_FREED
 @param[in,out]	mtr		mini-transaction
 @param[out]	block		inode block, or NULL to ignore
 @return segment inode, page x-latched; NULL if the inode is free */
@@ -1490,6 +1493,7 @@ fseg_inode_try_get(
 	const fseg_header_t*	header,
 	ulint			space,
 	ulint			zip_size,
+	ulint			mode,
 	mtr_t*			mtr,
 	buf_block_t**		block)
 {
@@ -1500,11 +1504,11 @@ fseg_inode_try_get(
 	inode_addr.boffset = mach_read_from_2(header + FSEG_HDR_OFFSET);
 	ut_ad(space == mach_read_from_4(header + FSEG_HDR_SPACE));
 
-	inode = fut_get_ptr(space, zip_size, inode_addr, RW_SX_LATCH, mtr,
-			    block);
-
-	if (UNIV_UNLIKELY(!mach_read_from_8(inode + FSEG_ID))) {
+	inode = fut_get_ptr(space, zip_size, inode_addr, RW_SX_LATCH, mode,
+			    mtr, block);
 
+	if (UNIV_UNLIKELY(!inode)) {
+	} else if (UNIV_UNLIKELY(!mach_read_from_8(inode + FSEG_ID))) {
 		inode = NULL;
 	} else {
 		ut_ad(mach_read_from_4(inode + FSEG_MAGIC_N)
@@ -1530,10 +1534,10 @@ fseg_inode_get(
 	mtr_t*			mtr,
 	buf_block_t**		block = NULL)
 {
-	fseg_inode_t*	inode
-		= fseg_inode_try_get(header, space, zip_size, mtr, block);
-	ut_a(inode);
-	return(inode);
+  fseg_inode_t *inode= fseg_inode_try_get(header, space, zip_size, BUF_GET,
+                                          mtr, block);
+  ut_a(inode);
+  return inode;
 }
 
 /** Get the page number from the nth fragment page slot.
@@ -1891,7 +1895,8 @@ fseg_alloc_free_extent(
 
 		first = flst_get_first(inode + FSEG_FREE);
 
-		descr = xdes_lst_get_descriptor(space, first, xdes, mtr);
+		descr = xdes_lst_get_descriptor(space, first, BUF_GET, xdes,
+						mtr);
 	} else {
 		/* Segment free list was empty, allocate from space */
 		descr = fsp_alloc_free_extent(space, 0, xdes, mtr);
@@ -2077,7 +2082,8 @@ take_hinted_page:
 			return(NULL);
 		}
 
-		ret_descr = xdes_lst_get_descriptor(space, first, &xdes, mtr);
+		ret_descr = xdes_lst_get_descriptor(space, first, BUF_GET,
+						    &xdes, mtr);
 		ret_page = xdes_find_free(ret_descr);
 		if (ret_page == FIL_NULL) {
 			ut_ad(!has_done_reservation);
@@ -2536,31 +2542,26 @@ fseg_free_page_low(
 @param[in,out]	space		tablespace
 @param[in]	offset		page number
 @param[in,out]	mtr		mini-transaction */
-void
-fseg_free_page(
-	fseg_header_t*	seg_header,
-	fil_space_t*	space,
-	uint32_t	offset,
-	mtr_t*		mtr)
+void fseg_free_page(fseg_header_t *seg_header, fil_space_t *space,
+                    uint32_t offset, mtr_t *mtr)
 {
-	DBUG_ENTER("fseg_free_page");
-	fseg_inode_t*		seg_inode;
-	buf_block_t*		iblock;
-	mtr_x_lock_space(space, mtr);
+  DBUG_ENTER("fseg_free_page");
+  buf_block_t *iblock;
+  mtr_x_lock_space(space, mtr);
 
-	DBUG_LOG("fseg_free_page", "space_id: " << space->id
-		 << ", page_no: " << offset);
-
-	seg_inode = fseg_inode_get(seg_header, space->id, space->zip_size(),
-				   mtr,
-				   &iblock);
-	if (!space->full_crc32()) {
-		fil_block_check_type(*iblock, FIL_PAGE_INODE, mtr);
-	}
+  DBUG_PRINT("fseg_free_page",
+             ("space_id: " ULINTPF ", page_no: %u", space->id, offset));
 
-	fseg_free_page_low(seg_inode, iblock, space, offset, mtr);
+  if (fseg_inode_t *seg_inode= fseg_inode_try_get(seg_header,
+                                                  space->id, space->zip_size(),
+                                                  BUF_GET, mtr, &iblock))
+  {
+    if (!space->full_crc32())
+      fil_block_check_type(*iblock, FIL_PAGE_INODE, mtr);
+    fseg_free_page_low(seg_inode, iblock, space, offset, mtr);
+  }
 
-	DBUG_VOID_RETURN;
+  DBUG_VOID_RETURN;
 }
 
 /** Determine whether a page is free.
@@ -2682,11 +2683,15 @@ fseg_free_step(
 	ut_a(!xdes_is_free(descr, header_page % FSP_EXTENT_SIZE));
 	buf_block_t* iblock;
 	const ulint zip_size = space->zip_size();
-	inode = fseg_inode_try_get(header, space_id, zip_size, mtr, &iblock);
+	inode = fseg_inode_try_get(header, space_id, zip_size,
+				   BUF_GET_POSSIBLY_FREED, mtr, &iblock);
+	if (space->is_stopping()) {
+		DBUG_RETURN(true);
+	}
 
 	if (inode == NULL) {
-		ib::info() << "Double free of inode from "
-			<< page_id_t(space_id, header_page);
+		ib::warn() << "Double free of inode from "
+			   << page_id_t(space_id, header_page);
 		DBUG_RETURN(true);
 	}
 
@@ -2695,6 +2700,10 @@ fseg_free_step(
 	}
 	descr = fseg_get_first_extent(inode, space, mtr);
 
+	if (space->is_stopping()) {
+		DBUG_RETURN(true);
+	}
+
 	if (descr != NULL) {
 		/* Free the extent held by the segment */
 		fseg_free_extent(inode, iblock, space, xdes_get_offset(descr),
@@ -2740,8 +2749,6 @@ fseg_free_step_not_header(
 				the first fragment page of the segment */
 	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
-	ulint		n;
-	xdes_t*		descr;
 	fseg_inode_t*	inode;
 
 	const uint32_t space_id = page_get_space_id(page_align(header));
@@ -2750,15 +2757,24 @@ fseg_free_step_not_header(
 	fil_space_t*		space = mtr_x_lock_space(space_id, mtr);
 	buf_block_t*		iblock;
 
-	inode = fseg_inode_get(header, space_id, space->zip_size(), mtr,
-			       &iblock);
+	inode = fseg_inode_try_get(header, space_id, space->zip_size(),
+				   BUF_GET_POSSIBLY_FREED, mtr, &iblock);
+	if (space->is_stopping()) {
+		return true;
+	}
+
+	if (!inode) {
+		ib::warn() << "Double free of "
+			   << page_id_t(space_id,
+					page_get_page_no(page_align(header)));
+		return true;
+	}
+
 	if (!space->full_crc32()) {
 		fil_block_check_type(*iblock, FIL_PAGE_INODE, mtr);
 	}
 
-	descr = fseg_get_first_extent(inode, space, mtr);
-
-	if (descr != NULL) {
+	if (xdes_t* descr = fseg_get_first_extent(inode, space, mtr)) {
 		/* Free the extent held by the segment */
 		fseg_free_extent(inode, iblock, space, xdes_get_offset(descr),
 				 mtr);
@@ -2767,7 +2783,7 @@ fseg_free_step_not_header(
 
 	/* Free a frag page */
 
-	n = fseg_find_last_used_frag_page_slot(inode);
+	ulint n = fseg_find_last_used_frag_page_slot(inode);
 
 	ut_a(n != ULINT_UNDEFINED);
 
@@ -2817,7 +2833,8 @@ fseg_get_first_extent(
 	buf_block_t *xdes;
 
 	return(first.page == FIL_NULL ? NULL
-	       : xdes_lst_get_descriptor(space, first, &xdes, mtr));
+	       : xdes_lst_get_descriptor(space, first, BUF_GET_POSSIBLY_FREED,
+					 &xdes, mtr));
 }
 
 #ifdef UNIV_BTR_PRINT
diff --git a/storage/innobase/fut/fut0lst.cc b/storage/innobase/fut/fut0lst.cc
index e084f0b7935..122d31c1f6d 100644
--- a/storage/innobase/fut/fut0lst.cc
+++ b/storage/innobase/fut/fut0lst.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2019, 2020, MariaDB Corporation.
+Copyright (c) 2019, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -151,7 +151,8 @@ static void flst_insert_after(buf_block_t *base, uint16_t boffset,
   {
     buf_block_t *block;
     flst_node_t *next= fut_get_ptr(add->page.id().space(), add->zip_size(),
-                                   next_addr, RW_SX_LATCH, mtr, &block);
+                                   next_addr, RW_SX_LATCH, BUF_GET, mtr,
+                                   &block);
     flst_write_addr(*block, next + FLST_PREV,
                     add->page.id().page_no(), aoffset, mtr);
   }
@@ -202,7 +203,8 @@ static void flst_insert_before(buf_block_t *base, uint16_t boffset,
   {
     buf_block_t *block;
     flst_node_t *prev= fut_get_ptr(add->page.id().space(), add->zip_size(),
-                                   prev_addr, RW_SX_LATCH, mtr, &block);
+                                   prev_addr, RW_SX_LATCH, BUF_GET, mtr,
+                                   &block);
     flst_write_addr(*block, prev + FLST_NEXT,
                     add->page.id().page_no(), aoffset, mtr);
   }
@@ -253,7 +255,7 @@ void flst_add_last(buf_block_t *base, uint16_t boffset,
     const flst_node_t *c= addr.page == add->page.id().page_no()
       ? add->frame + addr.boffset
       : fut_get_ptr(add->page.id().space(), add->zip_size(), addr,
-                    RW_SX_LATCH, mtr, &cur);
+                    RW_SX_LATCH, BUF_GET, mtr, &cur);
     flst_insert_after(base, boffset, cur,
                       static_cast<uint16_t>(c - cur->frame),
                       add, aoffset, mtr);
@@ -286,7 +288,7 @@ void flst_add_first(buf_block_t *base, uint16_t boffset,
     const flst_node_t *c= addr.page == add->page.id().page_no()
       ? add->frame + addr.boffset
       : fut_get_ptr(add->page.id().space(), add->zip_size(), addr,
-                    RW_SX_LATCH, mtr, &cur);
+                    RW_SX_LATCH, BUF_GET, mtr, &cur);
     flst_insert_before(base, boffset, cur,
                        static_cast<uint16_t>(c - cur->frame),
                        add, aoffset, mtr);
@@ -321,9 +323,10 @@ void flst_remove(buf_block_t *base, uint16_t boffset,
     flst_node_t *prev= prev_addr.page == cur->page.id().page_no()
       ? cur->frame + prev_addr.boffset
       : fut_get_ptr(cur->page.id().space(), cur->zip_size(), prev_addr,
-                    RW_SX_LATCH, mtr, &block);
-    flst_write_addr(*block, prev + FLST_NEXT,
-                    next_addr.page, next_addr.boffset, mtr);
+                    RW_SX_LATCH, BUF_GET_POSSIBLY_FREED, mtr, &block);
+    if (prev)
+      flst_write_addr(*block, prev + FLST_NEXT,
+                      next_addr.page, next_addr.boffset, mtr);
   }
 
   if (next_addr.page == FIL_NULL)
@@ -335,9 +338,10 @@ void flst_remove(buf_block_t *base, uint16_t boffset,
     flst_node_t *next= next_addr.page == cur->page.id().page_no()
       ? cur->frame + next_addr.boffset
       : fut_get_ptr(cur->page.id().space(), cur->zip_size(), next_addr,
-                    RW_SX_LATCH, mtr, &block);
-    flst_write_addr(*block, next + FLST_PREV,
-                    prev_addr.page, prev_addr.boffset, mtr);
+                    RW_SX_LATCH, BUF_GET_POSSIBLY_FREED, mtr, &block);
+    if (next)
+      flst_write_addr(*block, next + FLST_PREV,
+                      prev_addr.page, prev_addr.boffset, mtr);
   }
 
   byte *len= &base->frame[boffset + FLST_LEN];
@@ -368,7 +372,7 @@ void flst_validate(const buf_block_t *base, uint16_t boffset, mtr_t *mtr)
     mtr2.start();
     const flst_node_t *node= fut_get_ptr(base->page.id().space(),
                                          base->zip_size(), addr,
-                                         RW_SX_LATCH, &mtr2);
+                                         RW_SX_LATCH, BUF_GET, &mtr2);
     addr= flst_get_next_addr(node);
     mtr2.commit();
   }
@@ -382,7 +386,7 @@ void flst_validate(const buf_block_t *base, uint16_t boffset, mtr_t *mtr)
     mtr2.start();
     const flst_node_t *node= fut_get_ptr(base->page.id().space(),
                                          base->zip_size(), addr,
-                                         RW_SX_LATCH, &mtr2);
+                                         RW_SX_LATCH, BUF_GET, &mtr2);
     addr= flst_get_prev_addr(node);
     mtr2.commit();
   }
diff --git a/storage/innobase/include/fut0fut.h b/storage/innobase/include/fut0fut.h
index a52fc256efa..07f3e6e2006 100644
--- a/storage/innobase/include/fut0fut.h
+++ b/storage/innobase/include/fut0fut.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2019, MariaDB Corporation.
+Copyright (c) 2019, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -35,6 +35,7 @@ Created 12/13/1995 Heikki Tuuri
 @param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
 @param[in]	addr		file address
 @param[in]	rw_latch	RW_S_LATCH, RW_X_LATCH, RW_SX_LATCH
+@param[in]	mode		BUF_GET or BUF_GET_POSSIBLY_FREED
 @param[out]	ptr_block	file page
 @param[in,out]	mtr		mini-transaction
 @return pointer to a byte in (*ptr_block)->frame; the *ptr_block is
@@ -46,6 +47,7 @@ fut_get_ptr(
 	ulint			zip_size,
 	fil_addr_t		addr,
 	rw_lock_type_t		rw_latch,
+	ulint			mode,
 	mtr_t*			mtr,
 	buf_block_t**		ptr_block = NULL)
 {
@@ -57,12 +59,18 @@ fut_get_ptr(
 	      || (rw_latch == RW_X_LATCH)
 	      || (rw_latch == RW_SX_LATCH));
 
-	block = buf_page_get(page_id_t(space, addr.page), zip_size,
-			     rw_latch, mtr);
-
-	ptr = buf_block_get_frame(block) + addr.boffset;
-
-	buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
+	block = buf_page_get_gen(page_id_t(space, addr.page), zip_size,
+				 rw_latch, NULL, mode, __FILE__, __LINE__,
+				 mtr);
+	if (!block) {
+	} else if (mode == BUF_GET_POSSIBLY_FREED
+		   && block->page.status == buf_page_t::FREED) {
+		block = NULL;
+	} else {
+		ptr = buf_block_get_frame(block) + addr.boffset;
+
+		buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
+	}
 
 	if (ptr_block != NULL) {
 		*ptr_block = block;