MDEV-31234 InnoDB does not free UNDO after the fix of MDEV-30671

trx_purge_truncate_history(): Only call trx_purge_truncate_rseg_history()
if the rollback segment is safe to process. This will avoid leaking undo
log pages that are not yet ready to be processed. This fixes a regression
that was introduced in
commit 0de3be8c (MDEV-30671).

trx_sys_t::any_active_transactions(): Separately count XA PREPARE
transactions.

srv_purge_should_exit(): Terminate slow shutdown if the history size
does not change and XA PREPARE transactions exist in the system.
This will avoid a hang of the test innodb.recovery_shutdown.

Tested by: Matthias Leich

storage/innobase/include/trx0sys.h

@@ -1055,7 +1055,7 @@ class trx_sys_t
   void close();
 
   /** @return total number of active (non-prepared) transactions */
-  ulint any_active_transactions();
+  size_t any_active_transactions(size_t *prepared= nullptr);
 
 
   /**

storage/innobase/srv/srv0srv.cc

@@ -1697,7 +1697,7 @@ void srv_master_callback(void*)
 }
 
 /** @return whether purge should exit due to shutdown */
-static bool srv_purge_should_exit()
+static bool srv_purge_should_exit(size_t old_history_size)
 {
   ut_ad(srv_shutdown_state <= SRV_SHUTDOWN_CLEANUP);
 
@@ -1708,7 +1708,12 @@ static bool srv_purge_should_exit()
     return true;
 
   /* Slow shutdown was requested. */
-  if (const size_t history_size= trx_sys.rseg_history_len)
+  size_t prepared, active= trx_sys.any_active_transactions(&prepared);
+  const size_t history_size= trx_sys.rseg_history_len;
+
+  if (!history_size);
+  else if (!active && history_size == old_history_size && prepared);
+  else
   {
     static time_t progress_time;
     time_t now= time(NULL);
@@ -1725,7 +1730,7 @@ static bool srv_purge_should_exit()
     return false;
   }
 
-  return !trx_sys.any_active_transactions();
+  return !active;
 }
 
 /*********************************************************************//**
@@ -1845,7 +1850,7 @@ static size_t srv_do_purge(ulint* n_total_purged)
 
 		*n_total_purged += n_pages_purged;
 	} while (n_pages_purged > 0 && !purge_sys.paused()
-		 && !srv_purge_should_exit());
+		 && !srv_purge_should_exit(rseg_history_len));
 
 	return(rseg_history_len);
 }
@@ -1960,7 +1965,7 @@ static void purge_coordinator_callback_low()
     }
   }
   while ((purge_sys.enabled() && !purge_sys.paused()) ||
-         !srv_purge_should_exit());
+         !srv_purge_should_exit(trx_sys.rseg_history_len));
 }
 
 static void purge_coordinator_callback(void*)
@@ -2031,10 +2036,14 @@ ulint srv_get_task_queue_length()
 /** Shut down the purge threads. */
 void srv_purge_shutdown()
 {
-	if (purge_sys.enabled()) {
+  if (purge_sys.enabled())
+  {
     if (!srv_fast_shutdown && !opt_bootstrap)
       srv_update_purge_thread_count(innodb_purge_threads_MAX);
-		while(!srv_purge_should_exit()) {
+    size_t history_size= trx_sys.rseg_history_len;
+    while (!srv_purge_should_exit(history_size))
+    {
+      history_size= trx_sys.rseg_history_len;
       ut_a(!purge_sys.paused());
       srv_wake_purge_thread_if_not_active();
       purge_coordinator_task.wait();

storage/innobase/trx/trx0purge.cc

@@ -448,12 +448,7 @@ trx_purge_truncate_rseg_history(
 	prev_hdr_addr.boffset = static_cast<uint16_t>(prev_hdr_addr.boffset
 						      - TRX_UNDO_HISTORY_NODE);
 
-	if (!rseg.trx_ref_count
-	    && rseg.needs_purge <= (purge_sys.head.trx_no
-				    ? purge_sys.head.trx_no
-				    : purge_sys.tail.trx_no)
-	    && mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_STATE
-				+ block->frame)
+	if (mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_STATE + block->frame)
 	    == TRX_UNDO_TO_PURGE
 	    && !mach_read_from_2(block->frame + hdr_addr.boffset
 				 + TRX_UNDO_NEXT_LOG)) {
@@ -544,6 +539,7 @@ static void trx_purge_truncate_history()
       ut_ad(rseg->id == i);
       ut_ad(rseg->is_persistent());
       mutex_enter(&rseg->mutex);
+      if (!rseg->trx_ref_count && rseg->needs_purge <= head.trx_no)
         trx_purge_truncate_rseg_history(*rseg, head);
       mutex_exit(&rseg->mutex);
     }

storage/innobase/trx/trx0sys.cc

@@ -325,15 +325,29 @@ trx_sys_t::close()
 }
 
 /** @return total number of active (non-prepared) transactions */
-ulint trx_sys_t::any_active_transactions()
+size_t trx_sys_t::any_active_transactions(size_t *prepared)
 {
-  uint32_t total_trx= 0;
-
-  trx_sys.trx_list.for_each([&total_trx](const trx_t &trx) {
-    if (trx.state == TRX_STATE_COMMITTED_IN_MEMORY ||
-        (trx.state == TRX_STATE_ACTIVE && trx.id))
+  size_t total_trx= 0, prepared_trx= 0;
+
+  trx_sys.trx_list.for_each([&](const trx_t &trx) {
+    switch (trx.state) {
+    case TRX_STATE_NOT_STARTED:
+      break;
+    case TRX_STATE_ACTIVE:
+      if (!trx.id)
+        break;
+      /* fall through */
+    case TRX_STATE_COMMITTED_IN_MEMORY:
       total_trx++;
+      break;
+    case TRX_STATE_PREPARED:
+    case TRX_STATE_PREPARED_RECOVERED:
+      prepared_trx++;
+    }
   });
 
+  if (prepared)
+    *prepared= prepared_trx;
+
   return total_trx;
 }