MDEV-15443 Invalid wsrep XID read from the rollback segmentbb-10.3-mdev15443

Problem:
=======
 Transaction with disabled wsrep_on variable stores only transaction id
in the rollback segment and it doesn't write any XID information in rollback
segment. While reading the latest wsrep checkpoint from InnoDB, it reads
the undefined wsrep xid from the rollback segment which has maximum transaction
id.

Solution:
=========
- Introduce new field TRX_RSEG_WSREP_XID_SEQNO in rollback segment header page
which keeps track of the latest wsrep information. Basically it stores
trx->no whenever the trasaction with wsrep_on variable is enabled.

- To fetch the latest binlog information, InnoDB compares the binlog filename
and binlog offset.

4 files changed, 92 insertions, 21 deletions

diff --git a/mysql-test/suite/galera/r/galera_wsrep_xid.result b/mysql-test/suite/galera/r/galera_wsrep_xid.result
new file mode 100644
index 00000000000..683ef362f6a
--- /dev/null
+++ b/mysql-test/suite/galera/r/galera_wsrep_xid.result
@@ -0,0 +1,14 @@
+CREATE TABLE t1 (f1 INT PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+connection node_2;
+SELECT * FROM t1;
+f1
+1
+SET SESSION wsrep_on=0;
+INSERT INTO t1 VALUES (2);
+DELETE FROM t1 WHERE f1 = 2;
+connection node_1;
+INSERT INTO t1 VALUES (2);
+connection node_2;
+connection node_1;
+DROP TABLE t1;
diff --git a/mysql-test/suite/galera/t/galera_wsrep_xid.test b/mysql-test/suite/galera/t/galera_wsrep_xid.test
new file mode 100644
index 00000000000..bbf582aa96c
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_wsrep_xid.test
@@ -0,0 +1,33 @@
+--source include/have_innodb.inc
+--source include/galera_cluster.inc
+
+# Initialize table on node_1
+CREATE TABLE t1 (f1 INT PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+
+# Go to node_2, verify that the previous INSERT completed.
+# Take node_2 out of the cluster, insert and delete a record
+# on a table with wsrep_on.
+--connection node_2
+SELECT * FROM t1;
+SET SESSION wsrep_on=0;
+INSERT INTO t1 VALUES (2);
+DELETE FROM t1 WHERE f1 = 2;
+
+# Shutdown node_2
+--source include/shutdown_mysqld.inc
+
+# On node_1, verify that the node has left the cluster.
+--connection node_1
+--let $wait_condition = SELECT VARIABLE_VALUE = 1 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_size';
+--source include/wait_condition.inc
+
+# Insert into t1 to enforce IST on node_2 when it is restarted.
+INSERT INTO t1 VALUES (2);
+
+# Restart node_2
+--connection node_2
+--source include/start_mysqld.inc
+
+--connection node_1
+DROP TABLE t1;
diff --git a/storage/innobase/include/trx0rseg.h b/storage/innobase/include/trx0rseg.h
index d68ece39911..f57b8e7fa06 100644
--- a/storage/innobase/include/trx0rseg.h
+++ b/storage/innobase/include/trx0rseg.h
@@ -245,15 +245,16 @@ If no binlog information is present, the first byte is NUL. */
 #ifdef WITH_WSREP
 /** The offset to WSREP XID headers */
 #define	TRX_RSEG_WSREP_XID_INFO		TRX_RSEG_MAX_TRX_ID + 16 + 512
-
+/** Sequence number of WSREP XID. It is used to find latest WSREP XID info. */
+#define	TRX_RSEG_WSREP_XID_SEQNO	TRX_RSEG_WSREP_XID_INFO
 /** WSREP XID format (1 if present and valid, 0 if not present) */
-#define TRX_RSEG_WSREP_XID_FORMAT	TRX_RSEG_WSREP_XID_INFO
+#define TRX_RSEG_WSREP_XID_FORMAT	TRX_RSEG_WSREP_XID_INFO + 8
 /** WSREP XID GTRID length */
-#define TRX_RSEG_WSREP_XID_GTRID_LEN	TRX_RSEG_WSREP_XID_INFO + 4
+#define TRX_RSEG_WSREP_XID_GTRID_LEN	TRX_RSEG_WSREP_XID_INFO + 12
 /** WSREP XID bqual length */
-#define TRX_RSEG_WSREP_XID_BQUAL_LEN	TRX_RSEG_WSREP_XID_INFO + 8
+#define TRX_RSEG_WSREP_XID_BQUAL_LEN	TRX_RSEG_WSREP_XID_INFO + 16
 /** WSREP XID data (XIDDATASIZE bytes) */
-#define TRX_RSEG_WSREP_XID_DATA		TRX_RSEG_WSREP_XID_INFO + 12
+#define TRX_RSEG_WSREP_XID_DATA		TRX_RSEG_WSREP_XID_INFO + 20
 #endif /* WITH_WSREP*/
 
 /*-------------------------------------------------------------*/
diff --git a/storage/innobase/trx/trx0rseg.cc b/storage/innobase/trx/trx0rseg.cc
index b9ca2481681..9932977f37b 100644
--- a/storage/innobase/trx/trx0rseg.cc
+++ b/storage/innobase/trx/trx0rseg.cc
@@ -83,6 +83,9 @@ trx_rseg_update_wsrep_checkpoint(
 	trx_sys_cur_xid_seqno = xid_seqno;
 #endif /* UNIV_DEBUG */
 
+	mlog_write_ull(TRX_RSEG_WSREP_XID_SEQNO + rseg_header,
+		       trx_sys.get_max_trx_id(), mtr);
+
 	mlog_write_ulint(TRX_RSEG_WSREP_XID_FORMAT + rseg_header,
 			 uint32_t(xid->formatID),
 			 MLOG_4BYTES, mtr);
@@ -209,7 +212,7 @@ bool trx_rseg_read_wsrep_checkpoint(XID& xid)
 		}
 
 		trx_id_t id = mach_read_from_8(rseg_header
-					       + TRX_RSEG_MAX_TRX_ID);
+					       + TRX_RSEG_WSREP_XID_SEQNO);
 
 		if (id < max_id) {
 			continue;
@@ -398,44 +401,60 @@ trx_undo_lists_init(trx_rseg_t* rseg, trx_id_t& max_trx_id,
 }
 
 /** Restore the state of a persistent rollback segment.
-@param[in,out]	rseg		persistent rollback segment
-@param[in,out]	max_trx_id	maximum observed transaction identifier
-@param[in,out]	max_rseg_trx_id	maximum observed TRX_RSEG_MAX_TRX_ID
+@param[in,out]	rseg			persistent rollback segment
+@param[in,out]	max_trx_id		maximum observed transaction identifier
+@param[in,out]	max_rseg_xid_seqno	maximum observed sequence number for
+					WSREP XID info
 @param[in,out]	mtr		mini-transaction */
 static
 void
 trx_rseg_mem_restore(
 	trx_rseg_t*	rseg,
 	trx_id_t&	max_trx_id,
-	trx_id_t&	max_rseg_trx_id,
+#ifdef	WITH_WSREP
+	trx_id_t&	max_rseg_xid_seqno,
+#endif
 	mtr_t*		mtr)
 {
 	trx_rsegf_t*	rseg_header = trx_rsegf_get_new(
 		rseg->space, rseg->page_no, mtr);
+	lint		binlog_offset;
 
 	if (mach_read_from_4(rseg_header + TRX_RSEG_FORMAT) == 0) {
 		trx_id_t id = mach_read_from_8(rseg_header
 					       + TRX_RSEG_MAX_TRX_ID);
-
+#ifdef 	WITH_WSREP
+		trx_id_t xid_seqno = mach_read_from_8(rseg_header
+						   + TRX_RSEG_WSREP_XID_SEQNO);
+#endif /* WITH_WSRSEP */
 		if (id > max_trx_id) {
 			max_trx_id = id;
 		}
 
-		if (id > max_rseg_trx_id) {
-			max_rseg_trx_id = id;
+		if (rseg_header[TRX_RSEG_BINLOG_NAME]) {
+
+			binlog_offset = mach_read_from_8(
+				rseg_header + TRX_RSEG_BINLOG_OFFSET);
+
+			if (strncmp((char*) rseg_header + TRX_RSEG_BINLOG_NAME,
+				    trx_sys.recovered_binlog_filename,
+				    TRX_RSEG_BINLOG_NAME_LEN) >= 0
+			    && (trx_sys.recovered_binlog_offset
+				< binlog_offset)) {
 
-			if (rseg_header[TRX_RSEG_BINLOG_NAME]) {
 				memcpy(trx_sys.recovered_binlog_filename,
 				       rseg_header + TRX_RSEG_BINLOG_NAME,
 				       TRX_RSEG_BINLOG_NAME_LEN);
-				trx_sys.recovered_binlog_offset = mach_read_from_8(
-						rseg_header
-						+ TRX_RSEG_BINLOG_OFFSET);
+				trx_sys.recovered_binlog_offset =
+					binlog_offset;
 			}
 
 #ifdef WITH_WSREP
-			trx_rseg_read_wsrep_checkpoint(
-				rseg_header, trx_sys.recovered_wsrep_xid);
+			if (xid_seqno > max_rseg_xid_seqno) {
+				max_rseg_xid_seqno = xid_seqno;
+				trx_rseg_read_wsrep_checkpoint(
+					rseg_header, trx_sys.recovered_wsrep_xid);
+			}
 #endif
 		}
 	}
@@ -513,11 +532,12 @@ static void trx_rseg_init_binlog_info(const page_t* page)
 void
 trx_rseg_array_init()
 {
-	trx_id_t max_trx_id = 0, max_rseg_trx_id = 0;
+	trx_id_t max_trx_id = 0;
 
 	*trx_sys.recovered_binlog_filename = '\0';
 	trx_sys.recovered_binlog_offset = -1;
 #ifdef WITH_WSREP
+	trx_id_t max_rseg_xid_seqno = 0;
 	memset(&trx_sys.recovered_wsrep_xid, 0,
 	       sizeof trx_sys.recovered_wsrep_xid);
 	trx_sys.recovered_wsrep_xid.formatID = -1;
@@ -549,7 +569,10 @@ trx_rseg_array_init()
 				ut_ad(!trx_sys.rseg_array[rseg_id]);
 				trx_sys.rseg_array[rseg_id] = rseg;
 				trx_rseg_mem_restore(
-					rseg, max_trx_id, max_rseg_trx_id,
+					rseg, max_trx_id,
+#ifdef WITH_WSREP
+					max_rseg_xid_seqno,
+#endif
 					&mtr);
 			}
 		}