MDEV-22388 Corrupted undo log record leads to server crash

trx_undo_rec_copy(): Return nullptr if the undo record is corrupted.

trx_undo_rec_get_undo_no(): Define inline with the declaration.

trx_purge_dummy_rec: Replaced with a -1 pointer.

row_undo_rec_get(), UndorecApplier::apply_undo_rec(): Check
if trx_undo_rec_copy() returned nullptr.

trx_purge_get_next_rec(): Return nullptr upon encountering any
corruption, to signal the end of purge.

9 files changed, 51 insertions, 130 deletions

diff --git a/storage/innobase/CMakeLists.txt b/storage/innobase/CMakeLists.txt
index bb10160370a..cc1ec1e302a 100644
--- a/storage/innobase/CMakeLists.txt
+++ b/storage/innobase/CMakeLists.txt
@@ -236,7 +236,6 @@ SET(INNOBASE_SOURCES
 	include/trx0i_s.h
 	include/trx0purge.h
 	include/trx0rec.h
-	include/trx0rec.inl
 	include/trx0roll.h
 	include/trx0rseg.h
 	include/trx0sys.h
diff --git a/storage/innobase/include/trx0purge.h b/storage/innobase/include/trx0purge.h
index ef9111845a6..0896d9c79a1 100644
--- a/storage/innobase/include/trx0purge.h
+++ b/storage/innobase/include/trx0purge.h
@@ -33,10 +33,6 @@ Created 3/26/1996 Heikki Tuuri
 
 #include <queue>
 
-/** A dummy undo record used as a return value when we have a whole undo log
-which needs no purge */
-extern trx_undo_rec_t	trx_purge_dummy_rec;
-
 /** Prepend the history list with an undo log.
 Remove the undo log segment from the rseg slot if it is too big for reuse.
 @param[in]	trx		transaction
diff --git a/storage/innobase/include/trx0rec.h b/storage/innobase/include/trx0rec.h
index 8143fd419fa..838226601f8 100644
--- a/storage/innobase/include/trx0rec.h
+++ b/storage/innobase/include/trx0rec.h
@@ -24,8 +24,7 @@ Transaction undo log record
 Created 3/26/1996 Heikki Tuuri
 *******************************************************/
 
-#ifndef trx0rec_h
-#define trx0rec_h
+#pragma once
 
 #include "trx0types.h"
 #include "row0types.h"
@@ -37,29 +36,31 @@ Created 3/26/1996 Heikki Tuuri
 
 /***********************************************************************//**
 Copies the undo record to the heap.
-@return own: copy of undo log record */
-UNIV_INLINE
-trx_undo_rec_t*
-trx_undo_rec_copy(
-/*==============*/
-	const trx_undo_rec_t*	undo_rec,	/*!< in: undo log record */
-	mem_heap_t*		heap);		/*!< in: heap where copied */
-/**********************************************************************//**
-Reads the undo log record type.
-@return record type */
-UNIV_INLINE
-ulint
-trx_undo_rec_get_type(
-/*==================*/
-	const trx_undo_rec_t*	undo_rec);	/*!< in: undo log record */
+@param undo_rec   record in an undo log page
+@param heap       memory heap
+@return copy of undo_rec
+@retval nullptr if the undo log record is corrupted */
+inline trx_undo_rec_t* trx_undo_rec_copy(const trx_undo_rec_t *undo_rec,
+                                         mem_heap_t *heap)
+{
+  const size_t offset= ut_align_offset(undo_rec, srv_page_size);
+  const size_t end= mach_read_from_2(undo_rec);
+  if (end <= offset || end >= srv_page_size - FIL_PAGE_DATA_END)
+    return nullptr;
+  const size_t len= end - offset;
+  trx_undo_rec_t *rec= static_cast<trx_undo_rec_t*>
+    (mem_heap_dup(heap, undo_rec, len));
+  mach_write_to_2(rec, len);
+  return rec;
+}
+
 /**********************************************************************//**
 Reads the undo log record number.
 @return undo no */
-UNIV_INLINE
-undo_no_t
-trx_undo_rec_get_undo_no(
-/*=====================*/
-	const trx_undo_rec_t*	undo_rec);	/*!< in: undo log record */
+inline undo_no_t trx_undo_rec_get_undo_no(const trx_undo_rec_t *undo_rec)
+{
+  return mach_u64_read_much_compressed(undo_rec + 3);
+}
 
 /**********************************************************************//**
 Returns the start of the undo record data area. */
@@ -345,7 +346,3 @@ inline table_id_t trx_undo_rec_get_table_id(const trx_undo_rec_t *rec)
   mach_read_next_much_compressed(&rec);
   return mach_read_next_much_compressed(&rec);
 }
-
-#include "trx0rec.inl"
-
-#endif /* trx0rec_h */
diff --git a/storage/innobase/include/trx0rec.inl b/storage/innobase/include/trx0rec.inl
deleted file mode 100644
index 02244d68b6f..00000000000
--- a/storage/innobase/include/trx0rec.inl
+++ /dev/null
@@ -1,73 +0,0 @@
-/*****************************************************************************
-
-Copyright (c) 1996, 2014, Oracle and/or its affiliates. All Rights Reserved.
-
-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
-Foundation; version 2 of the License.
-
-This program is distributed in the hope that it will be useful, but WITHOUT
-ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
-
-You should have received a copy of the GNU General Public License along with
-this program; if not, write to the Free Software Foundation, Inc.,
-51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA
-
-*****************************************************************************/
-
-/**************************************************//**
-@file include/trx0rec.ic
-Transaction undo log record
-
-Created 3/26/1996 Heikki Tuuri
-*******************************************************/
-
-/**********************************************************************//**
-Reads from an undo log record the record type.
-@return record type */
-UNIV_INLINE
-ulint
-trx_undo_rec_get_type(
-/*==================*/
-	const trx_undo_rec_t*	undo_rec)	/*!< in: undo log record */
-{
-	return(mach_read_from_1(undo_rec + 2) & (TRX_UNDO_CMPL_INFO_MULT - 1));
-}
-
-/**********************************************************************//**
-Reads the undo log record number.
-@return undo no */
-UNIV_INLINE
-undo_no_t
-trx_undo_rec_get_undo_no(
-/*=====================*/
-	const trx_undo_rec_t*	undo_rec)	/*!< in: undo log record */
-{
-	const byte*	ptr;
-
-	ptr = undo_rec + 3;
-
-	return(mach_u64_read_much_compressed(ptr));
-}
-
-/***********************************************************************//**
-Copies the undo record to the heap.
-@return own: copy of undo log record */
-UNIV_INLINE
-trx_undo_rec_t*
-trx_undo_rec_copy(
-/*==============*/
-	const trx_undo_rec_t*	undo_rec,	/*!< in: undo log record */
-	mem_heap_t*		heap)		/*!< in: heap where copied */
-{
-	ulint		len;
-
-	len = mach_read_from_2(undo_rec)
-		- ut_align_offset(undo_rec, srv_page_size);
-	ut_ad(len < srv_page_size);
-	trx_undo_rec_t* rec = static_cast<trx_undo_rec_t*>(
-		mem_heap_dup(heap, undo_rec, len));
-	mach_write_to_2(rec, len);
-	return rec;
-}
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index abfa7c61c1f..a8cddd6575d 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -360,7 +360,7 @@ public:
   page_id_t get_page_id() const { return page_id; }
 
   /** Handle the DML undo log and apply it on online indexes */
-  void apply_undo_rec();
+  inline void apply_undo_rec();
 
   ~UndorecApplier()
   {
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 8a81a45ca12..47625b91f35 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -1114,7 +1114,7 @@ row_purge(
 	trx_undo_rec_t*	undo_rec,	/*!< in: record to purge */
 	que_thr_t*	thr)		/*!< in: query thread */
 {
-	if (undo_rec != &trx_purge_dummy_rec) {
+	if (undo_rec != reinterpret_cast<trx_undo_rec_t*>(-1)) {
 		bool	updated_extern;
 
 		while (row_purge_parse_undo_rec(
diff --git a/storage/innobase/row/row0undo.cc b/storage/innobase/row/row0undo.cc
index 9c41862e0b4..4d6d779eee6 100644
--- a/storage/innobase/row/row0undo.cc
+++ b/storage/innobase/row/row0undo.cc
@@ -342,7 +342,11 @@ static bool row_undo_rec_get(undo_node_t* node)
 					   node->heap);
 	mtr.commit();
 
-	switch (trx_undo_rec_get_type(node->undo_rec)) {
+	if (UNIV_UNLIKELY(!node->undo_rec)) {
+		return false;
+	}
+
+	switch (node->undo_rec[2] & (TRX_UNDO_CMPL_INFO_MULT - 1)) {
 	case TRX_UNDO_INSERT_METADATA:
 		/* This record type was introduced in MDEV-11369
 		instant ADD COLUMN, which was implemented after
@@ -356,13 +360,12 @@ static bool row_undo_rec_get(undo_node_t* node)
 	case TRX_UNDO_INSERT_REC:
 	case TRX_UNDO_EMPTY:
 		node->roll_ptr |= 1ULL << ROLL_PTR_INSERT_FLAG_POS;
-		node->state = undo == temp
+		node->state = is_temp
 			? UNDO_INSERT_TEMPORARY : UNDO_INSERT_PERSISTENT;
 		break;
 	default:
-		node->state = undo == temp
+		node->state = is_temp
 			? UNDO_UPDATE_TEMPORARY : UNDO_UPDATE_PERSISTENT;
-		break;
 	}
 
 	trx->undo_no = node->undo_no = trx_undo_rec_get_undo_no(
diff --git a/storage/innobase/trx/trx0purge.cc b/storage/innobase/trx/trx0purge.cc
index e37efa733e2..31153c8e966 100644
--- a/storage/innobase/trx/trx0purge.cc
+++ b/storage/innobase/trx/trx0purge.cc
@@ -55,10 +55,6 @@ ulong		srv_max_purge_lag_delay = 0;
 /** The global data structure coordinating a purge */
 purge_sys_t	purge_sys;
 
-/** A dummy undo record used as a return value when we have a whole undo log
-which needs no purge */
-trx_undo_rec_t	trx_purge_dummy_rec;
-
 #ifdef UNIV_DEBUG
 my_bool		srv_purge_view_update_only_debug;
 #endif /* UNIV_DEBUG */
@@ -1022,7 +1018,9 @@ TRANSACTIONAL_TARGET static void trx_purge_choose_next_log()
 
 /***********************************************************************//**
 Gets the next record to purge and updates the info in the purge system.
-@return copy of an undo log record or pointer to the dummy undo log record */
+@return copy of an undo log record
+@retval -1 if there is nothing to purge
+@retval nullptr on corruption */
 static
 trx_undo_rec_t*
 trx_purge_get_next_rec(
@@ -1048,11 +1046,10 @@ trx_purge_get_next_rec(
 		/* Look for the next undo log and record to purge */
 
 		trx_purge_choose_next_log();
-
-		return(&trx_purge_dummy_rec);
+		return reinterpret_cast<trx_undo_rec_t*>(-1);
 	}
 
-	mtr_start(&mtr);
+	mtr.start();
 
 	const buf_block_t* undo_page
 		= buf_page_get_gen(page_id, 0, RW_S_LATCH, nullptr,
@@ -1060,7 +1057,7 @@ trx_purge_get_next_rec(
 	if (UNIV_UNLIKELY(!undo_page)) {
 corrupted:
 		mtr.commit();
-		return &trx_purge_dummy_rec;
+		return nullptr;
 	}
 
 	const buf_block_t* rec2_page = undo_page;
@@ -1105,16 +1102,16 @@ corrupted:
 	trx_undo_rec_t*	rec_copy = trx_undo_rec_copy(undo_page->page.frame
 						     + offset, heap);
 
-	mtr_commit(&mtr);
-
-	return(rec_copy);
+	mtr.commit();
+	return rec_copy;
 }
 
 /********************************************************************//**
 Fetches the next undo log record from the history list to purge. It must be
 released with the corresponding release function.
-@return copy of an undo log record or pointer to trx_purge_dummy_rec,
-if the whole undo log can skipped in purge; NULL if none left */
+@return copy of an undo log record
+@retval -1 if the whole undo log can skipped in purge
+@retval nullptr if nothing is left, or on corruption */
 static MY_ATTRIBUTE((warn_unused_result))
 trx_undo_rec_t*
 trx_purge_fetch_next_rec(
@@ -1130,13 +1127,12 @@ trx_purge_fetch_next_rec(
 		if (!purge_sys.next_stored) {
 			DBUG_PRINT("ib_purge",
 				   ("no logs left in the history list"));
-			return(NULL);
+			return nullptr;
 		}
 	}
 
 	if (purge_sys.tail.trx_no >= purge_sys.low_limit_no()) {
-
-		return(NULL);
+		return nullptr;
 	}
 
 	/* fprintf(stderr, "Thread %lu purging trx %llu undo record %llu\n",
@@ -1152,7 +1148,7 @@ trx_purge_fetch_next_rec(
 	/* The following call will advance the stored values of the
 	purge iterator. */
 
-	return(trx_purge_get_next_rec(n_pages_handled, heap));
+	return trx_purge_get_next_rec(n_pages_handled, heap);
 }
 
 /** Run a purge batch.
@@ -1229,7 +1225,8 @@ trx_purge_attach_undo_recs(ulint n_purge_threads)
 
 		if (purge_rec.undo_rec == NULL) {
 			break;
-		} else if (purge_rec.undo_rec == &trx_purge_dummy_rec) {
+		} else if (purge_rec.undo_rec
+			   == reinterpret_cast<trx_undo_rec_t*>(-1)) {
 			continue;
 		}
 
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index 353d82c89cf..56bef29cc70 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -307,8 +307,10 @@ inline void UndorecApplier::assign_rec(const buf_block_t &block,
   this->undo_rec= trx_undo_rec_copy(block.page.frame + offset, heap);
 }
 
-void UndorecApplier::apply_undo_rec()
+inline void UndorecApplier::apply_undo_rec()
 {
+  if (!undo_rec)
+    return;
   bool updated_extern= false;
   undo_no_t undo_no= 0;
   table_id_t table_id= 0;