MDEV-24596 : Assertion `state_ == s_exec || state_ == s_quitting' failed in wsrep::client_state::disable_streamingbb-10.4-MDEV-24596

There were multiple problems here
* wsrep_trx_fragment_size should not be set when wsrep is disabled or provider is not loaded
* wsrep_trx_fragment_unit should not be set when wsrep is disabled or provider is not loaded
* wsrep_debug has no effect if wsrep is disabled or provider is not loaded
* wsrep_start_position should not be set when wsrep is disabled or provider is not loaded any other value than default
* wsrep_start_position should be changed only when we are joiner or initialized
* wsrep_start_position should be allowed to set only a value that exits, thus
we need to add error handling to wsrep_sst_complete

3 files changed, 87 insertions, 14 deletions

diff --git a/sql/wsrep_priv.h b/sql/wsrep_priv.h
index e480331ba65..fb8467adc9d 100644
--- a/sql/wsrep_priv.h
+++ b/sql/wsrep_priv.h
@@ -42,7 +42,7 @@ extern wsrep_seqno_t local_seqno;
 extern Wsrep_schema* wsrep_schema;
 
 // a helper function
-void wsrep_sst_received(THD*, const wsrep_uuid_t&, wsrep_seqno_t,
+bool wsrep_sst_received(THD*, const wsrep_uuid_t&, wsrep_seqno_t,
                         const void*, size_t);
 
 void wsrep_notify_status(enum wsrep::server_state::state status,
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index c024f08dd22..cadabd94e7a 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -308,12 +308,33 @@ bool wsrep_before_SE()
 }
 
 // Signal end of SST
-static void wsrep_sst_complete (THD*                thd,
-                                int const           rcode)
+static bool wsrep_sst_complete (THD*                thd,
+                                int const           rcode,
+                                wsrep::gtid const   sst_gtid)
 {
   Wsrep_client_service client_service(thd, thd->wsrep_cs());
-  Wsrep_server_state::instance().sst_received(client_service, rcode);
+  Wsrep_server_state& server_state= Wsrep_server_state::instance();
+  enum wsrep::server_state::state state= server_state.state();
+  bool failed= false;
+
+  if ((state == Wsrep_server_state::s_joiner ||
+       state == Wsrep_server_state::s_initialized))
+    Wsrep_server_state::instance().sst_received(client_service,
+       rcode);
+  else
+  {
+    char start_pos_buf[FN_REFLEN];
+    ssize_t len= wsrep::print_to_c_str(sst_gtid, start_pos_buf, FN_REFLEN-1);
+    start_pos_buf[len]='\0';
+    WSREP_ERROR("SST failed for position %s initialized %d server_state %s",
+                start_pos_buf,
+                server_state.is_initialized(),
+                wsrep::to_c_string(state));
+    failed= true;
+  }
+
   wsrep_joiner_monitor_end();
+  return failed;
 }
 
   /*
@@ -325,13 +346,15 @@ static void wsrep_sst_complete (THD*                thd,
   @param seqno     [IN]               Initial state sequence number
   @param state     [IN]               Always NULL, also ignored by wsrep provider (?)
   @param state_len [IN]               Always 0, also ignored by wsrep provider (?)
+  @return true when successfull, false if error
 */
-void wsrep_sst_received (THD*                thd,
+bool wsrep_sst_received (THD*                thd,
                          const wsrep_uuid_t& uuid,
                          wsrep_seqno_t const seqno,
                          const void* const   state,
                          size_t const        state_len)
 {
+  bool error= false;
   /*
     To keep track of whether the local uuid:seqno should be updated. Also, note
     that local state (uuid:seqno) is updated/checkpointed only after we get an
@@ -371,8 +394,10 @@ void wsrep_sst_received (THD*                thd,
     if (WSREP_ON)
     {
       int const rcode(seqno < 0 ? seqno : 0);
-      wsrep_sst_complete(thd,rcode);
+      error= wsrep_sst_complete(thd,rcode, sst_gtid);
     }
+
+    return error;
 }
 
 static int sst_scan_uuid_seqno (const char* str,
@@ -653,7 +678,7 @@ err:
     /* Read committed isolation to avoid gap locking */
     thd->variables.tx_isolation= ISO_READ_COMMITTED;
 
-    wsrep_sst_complete (thd, -err);
+    wsrep_sst_complete (thd, -err, ret_gtid);
 
     delete thd;
     my_thread_end();
diff --git a/sql/wsrep_var.cc b/sql/wsrep_var.cc
index 727750ea794..cf78c028836 100644
--- a/sql/wsrep_var.cc
+++ b/sql/wsrep_var.cc
@@ -234,7 +234,7 @@ bool wsrep_set_local_position(THD* thd, const char* const value,
   wsrep_seqno_t const seqno= strtoll(value + uuid_len + 1, NULL, 10);
 
   if (sst) {
-    wsrep_sst_received (thd, uuid, seqno, NULL, 0);
+    return (wsrep_sst_received (thd, uuid, seqno, NULL, 0));
   } else {
     local_uuid= uuid;
     local_seqno= seqno;
@@ -258,15 +258,28 @@ bool wsrep_start_position_check (sys_var *self, THD* thd, set_var* var)
   // Verify the format.
   if (wsrep_start_position_verify(start_pos_buf)) return true;
 
+  // Give error if position is updated when wsrep is not enabled or
+  // provider is not loaded.
+  if ((!WSREP_ON || !Wsrep_server_state::instance().is_provider_loaded())
+      && strcmp(start_pos_buf, "00000000-0000-0000-0000-000000000000:-1"))
+  {
+    push_warning (thd, Sql_condition::WARN_LEVEL_WARN,
+                  ER_WRONG_VALUE_FOR_VAR,
+                  "Cannot set 'wsrep_start_position' because "
+                  "wsrep is switched off or provider is not loaded");
+    goto err;
+  }
+
   /*
     As part of further verification, we try to update the value and catch
-    errors (if any).
+    errors (if any) only when value actually has been changed.
   */
-  if (wsrep_set_local_position(thd, var->save_result.string_value.str,
+  if (strcmp(start_pos_buf, wsrep_start_position))
+  {
+    if (wsrep_set_local_position(thd, var->save_result.string_value.str,
                                var->save_result.string_value.length,
                                true))
-  {
-    goto err;
+      goto err;
   }
 
   return false;
@@ -472,8 +485,19 @@ bool wsrep_reject_queries_update(sys_var *self, THD* thd, enum_var_type type)
 
 bool wsrep_debug_update(sys_var *self, THD* thd, enum_var_type type)
 {
-    Wsrep_server_state::instance().debug_log_level(wsrep_debug);
-    return false;
+  // Give warnings if wsrep_debug is set and wsrep is disabled or
+  // provider is not loaded
+  if ((!WSREP_ON || !Wsrep_server_state::instance().is_provider_loaded())
+      && wsrep_debug)
+  {
+    push_warning (thd, Sql_condition::WARN_LEVEL_WARN,
+                  ER_WRONG_VALUE_FOR_VAR,
+                  "Setting 'wsrep_debug' has no effect because "
+                  "wsrep is switched off");
+  }
+
+  Wsrep_server_state::instance().debug_log_level(wsrep_debug);
+  return false;
 }
 
 static int wsrep_cluster_address_verify (const char* cluster_address_str)
@@ -745,6 +769,18 @@ bool wsrep_trx_fragment_size_update(sys_var* self, THD *thd, enum_var_type)
 {
   WSREP_DEBUG("wsrep_trx_fragment_size_update: %llu",
               thd->variables.wsrep_trx_fragment_size);
+
+  // Give error if wsrep_trx_fragment_size is set and wsrep is disabled or
+  // provider is not loaded
+  if (!WSREP_ON || !Wsrep_server_state::instance().is_provider_loaded())
+  {
+    push_warning (thd, Sql_condition::WARN_LEVEL_WARN,
+                  ER_WRONG_VALUE_FOR_VAR,
+                  "Cannot set 'wsrep_trx_fragment_size' because "
+                  "wsrep is switched off");
+    return true;
+  }
+
   if (thd->variables.wsrep_trx_fragment_size)
   {
     return thd->wsrep_cs().enable_streaming(
@@ -762,6 +798,18 @@ bool wsrep_trx_fragment_unit_update(sys_var* self, THD *thd, enum_var_type)
 {
   WSREP_DEBUG("wsrep_trx_fragment_unit_update: %lu",
               thd->variables.wsrep_trx_fragment_unit);
+
+  // Give error if wsrep_trx_fragment_unit is set and wsrep is disabled or
+  // provider is not loaded
+  if (!WSREP_ON || !Wsrep_server_state::instance().is_provider_loaded())
+  {
+    push_warning (thd, Sql_condition::WARN_LEVEL_WARN,
+                  ER_WRONG_VALUE_FOR_VAR,
+                  "Cannot set 'wsrep_trx_fragment_unit' because "
+                  "wsrep is switched off");
+    return true;
+  }
+
   if (thd->variables.wsrep_trx_fragment_size)
   {
     return thd->wsrep_cs().enable_streaming(