MDEV-33819 The purge of committed history is mis-parsing some log

In commit aa719b50 (part of MDEV-32050)
a bug was introduced in the function purge_sys_t::choose_next_log(),
which reimplements some logic that previously was part of
trx_purge_read_undo_rec(). We must invoke trx_undo_get_first_rec()
with the page number and offset of the undo log header, but we were
incorrectly invoking it on the current undo page number, which caused
us to parse undo records starting at an incorrect offset.

purge_sys_t::choose_next_log(): Pass the correct parameter to
trx_undo_page_get_first_rec().

trx_undo_page_get_next_rec(), trx_undo_page_get_first_rec(),
trx_undo_page_get_last_rec(): Add debug assertions and make the
code more robust by returning nullptr on corruption. Should we
detect any corrupted undo logs during the purge of committed
transaction history, the sanest thing to do is to pretend that
the end of an undo log was reached. If any garbage is left in
the tables, it will be ignored by anything else than
CHECK TABLE ... EXTENDED, and it can be removed by OPTIMIZE TABLE.

Thanks to Matthias Leich for providing an "rr replay" trace where
this bug could be found.

Reviewed by: Vladislav Lesin

storage/innobase/include/trx0undo.inl

@@ -125,5 +125,6 @@ trx_undo_page_get_next_rec(const buf_block_t *undo_page, uint16_t rec,
 {
   uint16_t end= trx_undo_page_get_end(undo_page, page_no, offset);
   uint16_t next= mach_read_from_2(undo_page->page.frame + rec);
-  return next == end ? nullptr : undo_page->page.frame + next;
+  ut_ad(next <= end);
+  return next >= end ? nullptr : undo_page->page.frame + next;
 }

storage/innobase/trx/trx0purge.cc

@@ -793,8 +793,7 @@ bool purge_sys_t::rseg_get_next_history_log()
     /* Read the previous log header. */
     trx_id_t trx_no= 0;
     if (const buf_block_t* undo_page=
-        get_page(page_id_t(rseg->space->id,
-                                     prev_log_addr.page)))
+        get_page(page_id_t(rseg->space->id, prev_log_addr.page)))
     {
       const byte *log_hdr= undo_page->page.frame + prev_log_addr.boffset;
       trx_no= mach_read_from_8(log_hdr + TRX_UNDO_TRX_NO);
@@ -887,7 +886,7 @@ bool purge_sys_t::choose_next_log()
       if (!b)
         goto purge_nothing;
       undo_rec=
-        trx_undo_page_get_first_rec(b, page_no, hdr_offset);
+        trx_undo_page_get_first_rec(b, hdr_page_no, hdr_offset);
       if (!undo_rec)
         goto purge_nothing;
     }

storage/innobase/trx/trx0undo.cc

@@ -134,8 +134,9 @@ trx_undo_page_get_first_rec(const buf_block_t *block, uint32_t page_no,
                             uint16_t offset)
 {
   uint16_t start= trx_undo_page_get_start(block, page_no, offset);
-  return start == trx_undo_page_get_end(block, page_no, offset)
-    ? nullptr : block->page.frame + start;
+  uint16_t end= trx_undo_page_get_end(block, page_no, offset);
+  ut_ad(start <= end);
+  return start >= end ? nullptr : block->page.frame + start;
 }
 
 /** Get the last undo log record on a page.
@@ -149,8 +150,10 @@ trx_undo_rec_t*
 trx_undo_page_get_last_rec(const buf_block_t *block, uint32_t page_no,
                            uint16_t offset)
 {
+  uint16_t start= trx_undo_page_get_start(block, page_no, offset);
   uint16_t end= trx_undo_page_get_end(block, page_no, offset);
-  return trx_undo_page_get_start(block, page_no, offset) == end
+  ut_ad(start <= end);
+  return start >= end
     ? nullptr
     : block->page.frame + mach_read_from_2(block->page.frame + end - 2);
 }