MDEV-32757 innodb_undo_log_truncate=ON is not crash safe

trx_purge_truncate_history(): Do not prematurely mark dirty pages
as clean. This will be done in mtr_t::commit_shrink() as part of
Shrink::operator()(mtr_memo_slot_t*). Also, register each dirty page
only once in the mini-transaction.

fsp_page_create(): Adjust and simplify the page creation during
undo tablespace truncation. We can directly reuse pages that are
already in buf_pool.page_hash.

This fixes a regression that was caused by
commit f5794e1d (MDEV-26445).

Tested by: Matthias Leich
Reviewed by: Thirunarayanan Balathandayuthapani

storage/innobase/fsp/fsp0fsp.cc

@@ -1054,7 +1054,7 @@ static
 buf_block_t*
 fsp_page_create(fil_space_t *space, page_no_t offset, mtr_t *mtr)
 {
-  buf_block_t *block, *free_block;
+  buf_block_t *block;
 
   if (UNIV_UNLIKELY(space->is_being_truncated))
   {
@@ -1063,26 +1063,48 @@ fsp_page_create(fil_space_t *space, page_no_t offset, mtr_t *mtr)
     mysql_mutex_lock(&buf_pool.mutex);
     block= reinterpret_cast<buf_block_t*>
       (buf_pool.page_hash_get_low(page_id, fold));
-    if (block && block->page.oldest_modification() <= 1)
-      block= nullptr;
+    if (!block)
+    {
+      mysql_mutex_unlock(&buf_pool.mutex);
+      goto create;
+    }
+
+    buf_block_buf_fix_inc(block, __FILE__, __LINE__);
     mysql_mutex_unlock(&buf_pool.mutex);
 
-    if (block)
+    ut_ad(block->page.state() == BUF_BLOCK_FILE_PAGE);
+    ut_ad(!block->page.ibuf_exist);
+#ifdef BTR_CUR_HASH_ADAPT
+    ut_ad(!block->index);
+#endif
+
+    if (mtr->have_x_latch(*block))
     {
+      buf_block_buf_fix_dec(block);
       ut_ad(block->page.buf_fix_count() >= 1);
       ut_ad(rw_lock_get_x_lock_count(&block->lock) == 1);
-      ut_ad(mtr->have_x_latch(*block));
-      free_block= block;
-      goto got_free_block;
+    }
+    else
+    {
+      rw_lock_x_lock(&block->lock);
+      if (UNIV_UNLIKELY(block->page.id() != page_id))
+      {
+        buf_block_buf_fix_dec(block);
+        rw_lock_x_unlock(&block->lock);
+        goto create;
+      }
+      mtr->memo_push(block, MTR_MEMO_PAGE_X_FIX);
     }
   }
-
-  free_block= buf_LRU_get_free_block(false);
-got_free_block:
-  block= buf_page_create(space, static_cast<uint32_t>(offset),
-                         space->zip_size(), mtr, free_block);
-  if (UNIV_UNLIKELY(block != free_block))
-    buf_pool.free_block(free_block);
+  else
+  {
+  create:
+    buf_block_t *free_block= buf_LRU_get_free_block(false);
+    block= buf_page_create(space, static_cast<uint32_t>(offset),
+                           space->zip_size(), mtr, free_block);
+    if (UNIV_UNLIKELY(block != free_block))
+      buf_pool.free_block(free_block);
+  }
 
   fsp_init_file_page(space, block, mtr);
   return block;

storage/innobase/trx/trx0purge.cc

@@ -659,13 +659,12 @@ void trx_purge_truncate_history()
 
       buf_page_t *prev= UT_LIST_GET_PREV(list, bpage);
 
-      if (bpage->id().space() == space.id &&
-          bpage->oldest_modification() != 1)
+      if (bpage->oldest_modification() > 2 &&
+          bpage->id().space() == space.id)
       {
         ut_ad(bpage->state() == BUF_BLOCK_FILE_PAGE);
         auto block= reinterpret_cast<buf_block_t*>(bpage);
-        block->fix();
-        ut_ad(rw_lock_s_lock_nowait(block->debug_latch, __FILE__, __LINE__));
+        buf_block_buf_fix_inc(block, __FILE__, __LINE__);
         buf_pool.flush_hp.set(prev);
         mysql_mutex_unlock(&buf_pool.flush_list_mutex);
 
@@ -676,15 +675,13 @@ void trx_purge_truncate_history()
         mysql_mutex_lock(&buf_pool.flush_list_mutex);
         ut_ad(bpage->io_fix() == BUF_IO_NONE);
 
-        if (bpage->oldest_modification() > 1)
-        {
-          bpage->clear_oldest_modification(false);
+        if (bpage->oldest_modification() > 2 &&
+            !mtr.have_x_latch(*reinterpret_cast<buf_block_t*>(bpage)))
           mtr.memo_push(block, MTR_MEMO_PAGE_X_FIX);
-        }
         else
         {
+          buf_block_buf_fix_dec(block);
           rw_lock_x_unlock(&block->lock);
-          block->unfix();
         }
 
         if (prev != buf_pool.flush_hp.get())