MDEV-21117: --tc-heuristic-recover=rollback is not replication safebb-10.1-andrei

Problem:
=======
When run after master server crash --tc-heuristic-recover=rollback produces
inconsistent server state with binlog still containing transactions that were
rolled back by the option. Such way recovered server may not be used for
replication.

Fix:
===
A new --tc-heuristic-recover=BINLOG_TRUNCATE value is added to the
server option.  It works as ROLLBACK for engines but makes its best to
synchronize binlog to truncate it to the last committed transaction.

The BINLOG_TRUNCATE algorithm starts scanning possibly a sequences of
binlog files starting from the binlog checkpoint file to find a
truncate position candidate.  The candidate truncate position
corresponds to the beginning of the first GTID event whose transaction
is not found to have been committed. There must be no committed
transactions beyond this position *but* non-transactional events are
possible. This is also checked to warn when the case.
Rollback and possibly completion of partial commits (in the multi-engine case)
is done regardless, as well as the binlog GTID state is adjusted
accordingly. Also a STOP binlog event is written at the end of
truncated binary log.

1 files changed, 40 insertions, 9 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index 146d8bd5dbe..ba254f784c6 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -55,6 +55,7 @@
 #define my_b_write_string(A, B) my_b_write((A), (uchar*)(B), (uint) (sizeof(B) - 1))
 
 using std::max;
+using std::min;
 
 /**
   BINLOG_CHECKSUM variable.
@@ -6703,10 +6704,13 @@ bool Binlog_checkpoint_log_event::write()
 
 Gtid_log_event::Gtid_log_event(const char *buf, uint event_len,
                const Format_description_log_event *description_event)
-  : Log_event(buf, description_event), seq_no(0), commit_id(0)
+  : Log_event(buf, description_event), seq_no(0), commit_id(0),
+    flags_extra(0), extra_engines(0)
 {
   uint8 header_size= description_event->common_header_len;
   uint8 post_header_len= description_event->post_header_len[GTID_EVENT-1];
+  const char *buf_0= buf;
+
   if (event_len < header_size + post_header_len ||
       post_header_len < GTID_HEADER_LEN)
     return;
@@ -6717,6 +6721,7 @@ Gtid_log_event::Gtid_log_event(const char *buf, uint event_len,
   domain_id= uint4korr(buf);
   buf+= 4;
   flags2= *buf;
+  ++buf;
   if (flags2 & FL_GROUP_COMMIT_ID)
   {
     if (event_len < (uint)header_size + GTID_HEADER_LEN + 2)
@@ -6724,9 +6729,26 @@ Gtid_log_event::Gtid_log_event(const char *buf, uint event_len,
       seq_no= 0;                                // So is_valid() returns false
       return;
     }
-    ++buf;
     commit_id= uint8korr(buf);
+    buf+= 8;
   }
+  // the extra flags check and actions
+  if (buf - buf_0 < event_len)
+  {
+    flags_extra= *buf;
+    ++buf;
+    /* extra flags presence is identifed by non-zero byte value at this point */
+    if (flags_extra & FL_EXTRA_MULTI_ENGINE)
+    {
+      extra_engines= uint4korr(buf);
+      buf += 4;
+
+      DBUG_ASSERT(extra_engines > 0);
+    }
+  }
+  /* the '<' part of the assert corresponds to zero-padded trailing bytes */
+  DBUG_ASSERT(buf - buf_0 <= event_len);
+  DBUG_ASSERT(buf - buf_0 == event_len || buf_0[event_len - 1] == 0);
 }
 
 
@@ -6738,12 +6760,13 @@ Gtid_log_event::Gtid_log_event(THD *thd_arg, uint64 seq_no_arg,
                                uint64 commit_id_arg)
   : Log_event(thd_arg, flags_arg, is_transactional),
     seq_no(seq_no_arg), commit_id(commit_id_arg), domain_id(domain_id_arg),
-    flags2((standalone ? FL_STANDALONE : 0) | (commit_id_arg ? FL_GROUP_COMMIT_ID : 0))
+    flags2((standalone ? FL_STANDALONE : 0) |
+           (commit_id_arg ? FL_GROUP_COMMIT_ID : 0)),
+    flags_extra(0)
 {
   cache_type= Log_event::EVENT_NO_CACHE;
   if (thd_arg->transaction.stmt.trans_did_wait() ||
-      thd_arg->transaction.all.trans_did_wait())
-    flags2|= FL_WAITED;
+      thd_arg->transaction.all.trans_did_wait())flags2|= FL_WAITED;
   if (thd_arg->transaction.stmt.trans_did_ddl() ||
       thd_arg->transaction.stmt.has_created_dropped_temp_table() ||
       thd_arg->transaction.all.trans_did_ddl() ||
@@ -6756,6 +6779,8 @@ Gtid_log_event::Gtid_log_event(THD *thd_arg, uint64 seq_no_arg,
   /* Preserve any DDL or WAITED flag in the slave's binlog. */
   if (thd_arg->rgi_slave)
     flags2|= (thd_arg->rgi_slave->gtid_ev_flags2 & (FL_DDL|FL_WAITED));
+  if ((extra_engines= (min<uint>(2, ha_count_rw_all(thd, NULL, true)) - 2) > 0))
+    flags_extra|= FL_EXTRA_MULTI_ENGINE;
 }
 
 
@@ -6798,8 +6823,8 @@ Gtid_log_event::peek(const char *event_start, size_t event_len,
 bool
 Gtid_log_event::write()
 {
-  uchar buf[GTID_HEADER_LEN+2];
-  size_t write_len;
+  uchar buf[GTID_HEADER_LEN+2 + /* flags_extra: */ 1+4];
+  size_t write_len= 13;
 
   int8store(buf, seq_no);
   int4store(buf+8, domain_id);
@@ -6809,9 +6834,15 @@ Gtid_log_event::write()
     int8store(buf+13, commit_id);
     write_len= GTID_HEADER_LEN + 2;
   }
-  else
+  if (flags & FL_EXTRA_MULTI_ENGINE)
+  {
+    int4store(buf, extra_engines);
+    write_len += 4;
+  }
+
+  if (write_len < GTID_HEADER_LEN)
   {
-    bzero(buf+13, GTID_HEADER_LEN-13);
+    bzero(buf+write_len, GTID_HEADER_LEN-write_len);
     write_len= GTID_HEADER_LEN;
   }
   return write_header(write_len) ||