MDEV-21469: Implement crash-safe binary logging of the user XAbb-10.5-MDEV_21469

Make XA PREPARE, XA COMMIT and XA ROLLBACK statements
crash-safe when --log-bin is specified.

At execution of XA PREPARE, XA COMMIT and XA ROLLBACK their replication
events are made written into the binary log prior to execution
of the commands in storage engines.
In case the server crashes, after writing to binary log but before all
involved engines have processed the command, the following recovery
will execute the command's replication events to equalize the states
of involved engines with that of binlog.
That applies to all XA PREPARE *group* and XA COMMIT or ROLLBACK.

On the implementation level the recovery time binary log parsing
is augmented to pay attention to
the user XA xids to identify the XA transactions' state:s in binary log, and
eventually match them against their states in engines, see
MYSQL_BIN_LOG::recover_explicit_xa_prepare().
In discrepancy cases the outdated state in the engines is corrected with
resubmitting the transaction prepare group of events, or completion
ones. The multi-engine partly prepared XA PREPARE case
the XA is rolled back first.
The fact of multiple-engine involved is registered into
Gtid_log_event::flags2 as one bit. The boolean value is
sufficient and precise to deal with two engines in XA transaction.
With more than 2 recoverable engines the flag method is still correct though
may be pessimistic, as it treats all recoverable engines as XA
participants. So when the number of such Engines exceeds the number
of prepared engines of the XA that XA is treated
as partially completed, with all that ensued.
As an optimization no new bit is allocated in flags2, instead a
pre-existing ones (of MDEV-742) are reused, observing that
A. XA "COMPLETE" does not require multi-engine hint for its recovery and that
B. the MDEV-742 XA-completion bit is not anyhow used by
   XA-PREPARE and its GTID log event.

Notice the multi-engine flagging is superceded by MDEV-21117 extension
in Gtid log event so this part should be taken from there.

1 files changed, 52 insertions, 13 deletions

diff --git a/sql/handler.cc b/sql/handler.cc
index ec14e6cbf95..b981294f0d6 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1345,6 +1345,9 @@ int ha_prepare(THD *thd)
       handlerton *ht= ha_info->ht();
       if (ht->prepare)
       {
+        DBUG_EXECUTE_IF("simulate_crash_after_first_engine_prepare",
+                        if (!ha_info->next()) DBUG_SUICIDE(););
+
         if (unlikely(prepare_or_error(ht, thd, all)))
         {
           ha_rollback_trans(thd, all);
@@ -1375,22 +1378,22 @@ int ha_prepare(THD *thd)
 }
 
 /*
-  Like ha_check_and_coalesce_trx_read_only to return counted number of
-  read-write transaction participants limited to two, but works in the 'all'
-  context.
-  Also returns the last found rw ha_info through the 2nd argument.
+  Returns counted number of
+  read-write recoverable transaction participants optionally limited to two.
+  Also optionally returns the last found rw ha_info through the 2nd argument.
 */
-uint ha_count_rw_all(THD *thd, Ha_trx_info **ptr_ha_info)
+uint ha_count_rw_all(THD *thd, Ha_trx_info **ptr_ha_info, bool count_through)
 {
   unsigned rw_ha_count= 0;
 
   for (auto ha_info= thd->transaction.all.ha_list; ha_info;
        ha_info= ha_info->next())
   {
-    if (ha_info->is_trx_read_write())
+    if (ha_info->is_trx_read_write() && ha_info->ht()->recover)
     {
-      *ptr_ha_info= ha_info;
-      if (++rw_ha_count > 1)
+      if (ptr_ha_info)
+        *ptr_ha_info= ha_info;
+      if (++rw_ha_count > 1 && !count_through)
         break;
     }
   }
@@ -1403,7 +1406,7 @@ uint ha_count_rw_all(THD *thd, Ha_trx_info **ptr_ha_info)
   A helper function to evaluate if two-phase commit is mandatory.
   As a side effect, propagates the read-only/read-write flags
   of the statement transaction to its enclosing normal transaction.
-  
+
   If we have at least two engines with read-write changes we must
   run a two-phase commit. Otherwise we can run several independent
   commits as the only transactional engine has read-write changes
@@ -1883,6 +1886,10 @@ commit_one_phase_2(THD *thd, bool all, THD_TRANS *trans, bool is_real_trans)
     {
       int err;
       handlerton *ht= ha_info->ht();
+
+      DBUG_EXECUTE_IF("simulate_crash_after_first_engine_commit_or_rollback",
+                      if (!ha_info->next()) DBUG_SUICIDE(););
+
       if ((err= ht->commit(ht, thd, all)))
       {
         my_error(ER_ERROR_DURING_COMMIT, MYF(0), err);
@@ -1994,6 +2001,10 @@ int ha_rollback_trans(THD *thd, bool all)
     {
       int err;
       handlerton *ht= ha_info->ht();
+
+      DBUG_EXECUTE_IF("simulate_crash_after_first_engine_commit_or_rollback",
+                      if (!ha_info->next()) DBUG_SUICIDE(););
+
       if ((err= ht->rollback(ht, thd, all)))
       { // cannot happen
         my_error(ER_ERROR_DURING_ROLLBACK, MYF(0), err);
@@ -2233,7 +2244,9 @@ struct xarecover_st
   int len, found_foreign_xids, found_my_xids;
   XID *list;
   HASH *commit_list;
+  HASH *xa_prepared_list; // prepared user xa list
   bool dry_run;
+  uint recover_htons;     // number of recoverable htons for XA recovery
 };
 
 static my_bool xarecover_handlerton(THD *unused, plugin_ref plugin,
@@ -2245,6 +2258,7 @@ static my_bool xarecover_handlerton(THD *unused, plugin_ref plugin,
 
   if (hton->recover)
   {
+    info->recover_htons++;
     while ((got= hton->recover(hton, info->list, info->len)) > 0 )
     {
       sql_print_information("Found %d prepared transaction(s) in %s",
@@ -2286,7 +2300,21 @@ static my_bool xarecover_handlerton(THD *unused, plugin_ref plugin,
             _db_doprnt_("ignore xid %s", xid_to_str(buf, info->list[i]));
             });
           xid_cache_insert(info->list + i);
+          XID *foreign_xid= info->list + i;
           info->found_foreign_xids++;
+
+           /*
+             For each foreign xid prepraed in engine, check if it is present in
+             xa_prepared_list of binlog.
+           */
+          if (info->xa_prepared_list)
+          {
+            struct xa_recovery_member *member= NULL;
+            if ((member= (xa_recovery_member *)
+                 my_hash_search(info->xa_prepared_list, foreign_xid->key(),
+                                foreign_xid->key_length())))
+              member->in_engine_prepare++;
+          }
           continue;
         }
         if (IF_WSREP(!(wsrep_emulate_bin_log &&
@@ -2333,14 +2361,23 @@ static my_bool xarecover_handlerton(THD *unused, plugin_ref plugin,
   return FALSE;
 }
 
-int ha_recover(HASH *commit_list)
+/*
+  The function accepts two xid record hashes of regular and user XA resp.
+  The regular transactions recovery is decided right here.
+  The user XA recovery will proceed. For that the function
+  update the states of the user xa xid records and also returns
+  the number of recoverable htons.
+*/
+int ha_recover(HASH *commit_list, HASH *xa_prepared_list, uint *ptr_count)
 {
   struct xarecover_st info;
   DBUG_ENTER("ha_recover");
   info.found_foreign_xids= info.found_my_xids= 0;
   info.commit_list= commit_list;
+  info.xa_prepared_list= xa_prepared_list;
   info.dry_run= (info.commit_list==0 && tc_heuristic_recover==0);
   info.list= NULL;
+  info.recover_htons= 0;
 
   /* commit_list and tc_heuristic_recover cannot be set both */
   DBUG_ASSERT(info.commit_list==0 || tc_heuristic_recover==0);
@@ -2367,12 +2404,14 @@ int ha_recover(HASH *commit_list)
     DBUG_RETURN(1);
   }
 
-  plugin_foreach(NULL, xarecover_handlerton, 
+  plugin_foreach(NULL, xarecover_handlerton,
                  MYSQL_STORAGE_ENGINE_PLUGIN, &info);
 
+  if (ptr_count)
+    *ptr_count= info.recover_htons;
   my_free(info.list);
   if (info.found_foreign_xids)
-    sql_print_warning("Found %d prepared XA transactions", 
+    sql_print_warning("Found %d prepared XA transactions",
                       info.found_foreign_xids);
   if (info.dry_run && info.found_my_xids)
   {
@@ -2385,7 +2424,7 @@ int ha_recover(HASH *commit_list)
                     info.found_my_xids, opt_tc_log_file);
     DBUG_RETURN(1);
   }
-  if (info.commit_list)
+  if (info.commit_list && !info.found_foreign_xids)
     sql_print_information("Crash recovery finished.");
   DBUG_RETURN(0);
 }