MDEV-32115: Log checkpoint race with wsrep_sst_method=rsync

Galera snapshot transfer (SST) using the default wsrep_sst_method=rsync
is prone to creating corrupted snapshots. The probability for this is
rather low and might only affect installations that include
ENGINE=InnoDB tables that contain FULLTEXT INDEX.

The function sst_disable_innodb_writes() aims to disable all InnoDB writes
during the time a snapshot transfer (SST) is in progress using the
default wsrep_sst_method=rsync.

The logic based on invoking log_make_checkpoint() almost works, except
for two things: We failed to ensure that fts_optimize_callback() has
stopped executing, and we did not block updates of the log checkpoint
header.

log_checkpoint_low(): Assert that writes to the log are allowed.

buf_flush_page_cleaner(): Do not try to advance the checkpoint while
wsrep_sst_method=rsync is in progress. This prevents the assertion
in log_checkpoint_low() from failing.

fts_optimize_pause(), fts_optimize_resume(): Pause and resume the
fts_optimize_callback().

sst_disable_innodb_writes(): Disable all background writers
before initiating the log checkpoint.

fts_optimize_callback(): Assert that wsrep_sst_method=rsync is not
active, and remove the previous incorrect attempt at fixing this race.

Author: dr-m

storage/innobase/buf/buf0flu.cc

@@ -1966,6 +1966,7 @@ inline void log_t::write_checkpoint(lsn_t checkpoint, lsn_t end_lsn) noexcept
 static void log_checkpoint_low(lsn_t oldest_lsn, lsn_t end_lsn) noexcept
 {
   ut_ad(!srv_read_only_mode);
+  ut_ad(!recv_no_log_write);
   ut_ad(log_sys.latch_have_wr());
   ut_ad(oldest_lsn <= end_lsn);
   ut_ad(end_lsn == log_sys.get_lsn());
@@ -2550,6 +2551,11 @@ static void buf_flush_page_cleaner() noexcept
       {
         if (recv_recovery_is_on())
           continue;
+#ifdef WITH_WSREP
+        extern Atomic_relaxed<bool> wsrep_sst_disable_writes;
+        if (UNIV_UNLIKELY(wsrep_sst_disable_writes))
+          continue; /* See sst_disable_innodb_writes() */
+#endif
         IF_DBUG(if (log_sys.last_checkpoint_lsn &&
                     srv_shutdown_state < SRV_SHUTDOWN_CLEANUP &&
                     (_db_keyword_(nullptr, "ib_log_checkpoint_avoid", 1) ||

storage/innobase/fts/fts0opt.cc

@@ -37,13 +37,6 @@ Completed 2011/7/10 Sunny and Jimmy Yang
 #include "zlib.h"
 #include "fts0opt.h"
 #include "fts0vlc.h"
-#include "wsrep.h"
-
-#ifdef WITH_WSREP
-extern Atomic_relaxed<bool> wsrep_sst_disable_writes;
-#else
-constexpr bool wsrep_sst_disable_writes= false;
-#endif
 
 /** The FTS optimize thread's work queue. */
 ib_wqueue_t* fts_optimize_wq;
@@ -52,7 +45,7 @@ static void timer_callback(void*);
 static tpool::timer* timer;
 
 static tpool::task_group task_group(1);
-static tpool::task task(fts_optimize_callback,0, &task_group);
+static tpool::waitable_task task(fts_optimize_callback,0, &task_group);
 
 /** FTS optimize thread, for MDL acquisition */
 static THD *fts_opt_thd;
@@ -230,7 +223,7 @@ ulong	fts_num_word_optimize;
 char	fts_enable_diag_print;
 
 /** ZLib compressed block size.*/
-static ulint FTS_ZIP_BLOCK_SIZE	= 1024;
+static constexpr ulint FTS_ZIP_BLOCK_SIZE = 1024;
 
 /** The amount of time optimizing in a single pass, in seconds. */
 static ulint fts_optimize_time_limit;
@@ -2831,6 +2824,10 @@ static void fts_optimize_callback(void *)
 	static ulint		n_tables = ib_vector_size(fts_slots);
 
 	while (!done && srv_shutdown_state <= SRV_SHUTDOWN_INITIATED) {
+#ifdef WITH_WSREP
+		ut_d(extern Atomic_relaxed<bool> wsrep_sst_disable_writes);
+		ut_ad(!wsrep_sst_disable_writes);
+#endif
 		/* If there is no message in the queue and we have tables
 		to optimize then optimize the tables. */
 
@@ -2841,17 +2838,6 @@ static void fts_optimize_callback(void *)
 
 			/* The queue is empty but we have tables
 			to optimize. */
-			if (UNIV_UNLIKELY(wsrep_sst_disable_writes)) {
-retry_later:
-				if (fts_is_sync_needed()) {
-					fts_need_sync = true;
-				}
-				if (n_tables) {
-					timer->set_time(5000, 0);
-				}
-				return;
-			}
-
 			fts_slot_t* slot = static_cast<fts_slot_t*>(
 				ib_vector_get(fts_slots, current));
 
@@ -2872,7 +2858,13 @@ static void fts_optimize_callback(void *)
 				(ib_wqueue_nowait(fts_optimize_wq));
 			/* Timeout ? */
 			if (!msg) {
-				goto retry_later;
+				if (fts_is_sync_needed()) {
+					fts_need_sync = true;
+				}
+				if (n_tables) {
+					timer->set_time(5000, 0);
+				}
+				return;
 			}
 
 			switch (msg->type) {
@@ -2898,11 +2890,6 @@ static void fts_optimize_callback(void *)
 				break;
 
 			case FTS_MSG_SYNC_TABLE:
-				if (UNIV_UNLIKELY(wsrep_sst_disable_writes)) {
-					add_msg(msg);
-					goto retry_later;
-				}
-
 				DBUG_EXECUTE_IF(
 					"fts_instrument_msg_sync_sleep",
 					std::this_thread::sleep_for(
@@ -2945,11 +2932,8 @@ static void fts_optimize_callback(void *)
 	ib::info() << "FTS optimize thread exiting.";
 }
 
-/**********************************************************************//**
-Startup the optimize thread and create the work queue. */
-void
-fts_optimize_init(void)
-/*===================*/
+/** Startup the optimize task and create the work queue. */
+void fts_optimize_init()
 {
 	mem_heap_t*	heap;
 	ib_alloc_t*     heap_alloc;
@@ -2993,9 +2977,8 @@ fts_optimize_init(void)
 	last_check_sync_time = time(NULL);
 }
 
-/** Shutdown fts optimize thread. */
-void
-fts_optimize_shutdown()
+/** Shut down the fts optimize thread. */
+void fts_optimize_shutdown()
 {
 	ut_ad(!srv_read_only_mode);
 
@@ -3004,7 +2987,7 @@ fts_optimize_shutdown()
 	dict_sys.freeze(SRW_LOCK_CALL);
 	mysql_mutex_lock(&fts_optimize_wq->mutex);
 	/* Tells FTS optimizer system that we are exiting from
-	optimizer thread, message send their after will not be
+	optimizer thread, messages sent thereafter will not be
 	processed */
 	fts_opt_start_shutdown = true;
 	dict_sys.unfreeze();
@@ -3034,6 +3017,26 @@ fts_optimize_shutdown()
 	timer = NULL;
 }
 
+#ifdef WITH_WSREP
+/** Pause the optimize subsystem. */
+void fts_optimize_pause()
+{
+  ut_ad(!srv_read_only_mode);
+  /* Prevent fts_optimize_callback() from being scheduled. */
+  timer->disarm();
+  /* Wait for any current fts_optimize_callback() to finish. */
+  task.wait();
+}
+
+/** Resume after fts_optimize_stop() */
+void fts_optimize_resume()
+{
+  /* Schedule fts_optimize_callback() immediately.
+  It will reschedule itself via the timer when needed. */
+  srv_thread_pool->submit_task(&task);
+}
+#endif
+
 /** Sync the table during commit phase
 @param[in]	table	table to be synced */
 void fts_sync_during_ddl(dict_table_t* table)

storage/innobase/handler/ha_innodb.cc

@@ -1779,9 +1779,10 @@ static void sst_disable_innodb_writes()
   fil_crypt_set_thread_cnt(0);
   srv_n_fil_crypt_threads= old_count;
 
-  wsrep_sst_disable_writes= true;
   dict_stats_shutdown();
+  fts_optimize_pause();
   purge_sys.stop();
+
   /* We are holding a global MDL thanks to FLUSH TABLES WITH READ LOCK.
 
   That will prevent any writes from arriving into InnoDB, but it will
@@ -1793,10 +1794,12 @@ static void sst_disable_innodb_writes()
   possible during the snapshot, and to guarantee that no crash
   recovery will be necessary when starting up on the snapshot. */
   log_make_checkpoint();
+  wsrep_sst_disable_writes= true;
   /* If any FILE_MODIFY records were written by the checkpoint, an
   extra write of a FILE_CHECKPOINT record could still be invoked by
-  buf_flush_page_cleaner(). Let us prevent that by invoking another
-  checkpoint (which will write the FILE_CHECKPOINT record). */
+  buf_flush_page_cleaner(). Let us ensure that the page cleaner
+  is idle and will observe our above assignment (not write anything
+  further to the log). */
   log_make_checkpoint();
   ut_d(recv_no_log_write= true);
   /* If this were not a no-op, an assertion would fail due to
@@ -1811,6 +1814,8 @@ static void sst_enable_innodb_writes()
   dict_stats_start();
   purge_sys.resume();
   wsrep_sst_disable_writes= false;
+  /* Allow fts_optimize_callback() to assert that the flag is clear. */
+  fts_optimize_resume();
   const uint old_count= srv_n_fil_crypt_threads;
   srv_n_fil_crypt_threads= 0;
   fil_crypt_set_thread_cnt(old_count);

storage/innobase/include/fts0fts.h

@@ -624,11 +624,8 @@ fts_optimize_table(
 /*===============*/
 	dict_table_t*	table);			/*!< in: table to optimiza */
 
-/**********************************************************************//**
-Startup the optimize thread and create the work queue. */
-void
-fts_optimize_init(void);
-/*====================*/
+/** Startup the optimize task and create the work queue. */
+void fts_optimize_init();
 
 /****************************************************************//**
 Drops index ancillary tables for a FTS index
@@ -651,8 +648,15 @@ fts_optimize_remove_table(
 	dict_table_t*	table);			/*!< in: table to remove */
 
 /** Shutdown fts optimize thread. */
-void
-fts_optimize_shutdown();
+void fts_optimize_shutdown();
+
+#ifdef WITH_WSREP
+/** Pause the optimize subsystem. */
+void fts_optimize_pause();
+
+/** Resume after fts_optimize_pause() */
+void fts_optimize_resume();
+#endif
 
 /** Send sync fts cache for the table.
 @param[in]	table	table to sync */