MDEV-6368: assertion xid_seqno > trx_sys_cur_xid_seqno

- Validate the specified wsrep_start_position value by also
checking the return status of wsrep->sst_received. This also
ensures that changes in wsrep_start_position is not allowed
when the node is not in JOINING state.
- Do not allow decrease in seqno within same UUID.
- The initial checkpoint in SEs should be [0...:-1].

9 files changed, 187 insertions, 77 deletions

diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 845d114bc7a..ad3c96fb4c3 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -5979,7 +5979,10 @@ int mysqld_main(int argc, char **argv)
         wsrep_SE_init_grab();
         wsrep_SE_init_done();
         /*! in case of SST wsrep waits for wsrep->sst_received */
-        wsrep_sst_continue();
+        if (wsrep_sst_continue())
+        {
+          WSREP_ERROR("Failed to signal the wsrep provider to continue.");
+        }
       }
       else
       {
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 3cdf95e2bf2..531b624fad1 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -556,8 +556,11 @@ int wsrep_init()
   int rcode= -1;
   DBUG_ASSERT(wsrep_inited == 0);
 
-  if (strcmp(wsrep_start_position, WSREP_START_POSITION_ZERO))
-    wsrep_start_position_init(wsrep_start_position);
+  if (strcmp(wsrep_start_position, WSREP_START_POSITION_ZERO) &&
+      wsrep_start_position_init(wsrep_start_position))
+  {
+    return 1;
+  }
 
   wsrep_sst_auth_init(wsrep_sst_auth);
 
diff --git a/sql/wsrep_priv.h b/sql/wsrep_priv.h
index 30dce78c1a4..a6e57e33e6d 100644
--- a/sql/wsrep_priv.h
+++ b/sql/wsrep_priv.h
@@ -40,8 +40,8 @@ extern wsrep_uuid_t  local_uuid;
 extern wsrep_seqno_t local_seqno;
 
 // a helper function
-void wsrep_sst_received(wsrep_t*, const wsrep_uuid_t&, wsrep_seqno_t,
-                        const void*, size_t);
+bool wsrep_sst_received(wsrep_t*, const wsrep_uuid_t&, wsrep_seqno_t,
+                        const void*, size_t, bool const);
 /*! SST thread signals init thread about sst completion */
 void wsrep_sst_complete(const wsrep_uuid_t*, wsrep_seqno_t, bool);
 
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index 3ebbedaa04c..4fb162f9995 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -264,52 +264,110 @@ void wsrep_sst_complete (const wsrep_uuid_t* sst_uuid,
   mysql_mutex_unlock (&LOCK_wsrep_sst);
 }
 
-void wsrep_sst_received (wsrep_t*            const wsrep,
+/*
+  If wsrep provider is loaded, inform that the new state snapshot
+  has been received. Also update the local checkpoint.
+
+  @param wsrep     [IN]               wsrep handle
+  @param uuid      [IN]               Initial state UUID
+  @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 (?)
+  @param implicit  [IN]               Whether invoked implicitly due to SST
+                                      (true) or explicitly because if change
+                                      in wsrep_start_position by user (false).
+  @return false                       Success
+          true                        Error
+
+*/
+bool wsrep_sst_received (wsrep_t*            const wsrep,
                          const wsrep_uuid_t&       uuid,
-                         wsrep_seqno_t       const seqno,
+                         const wsrep_seqno_t       seqno,
                          const void*         const state,
-                         size_t              const state_len)
+                         const size_t              state_len,
+                         const bool                implicit)
 {
-    wsrep_get_SE_checkpoint(local_uuid, local_seqno);
+  /*
+    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
+    OK from wsrep provider. By doing so, the values remain consistent across
+    the server & wsrep provider.
+  */
+  bool do_update= false;
 
-    if (memcmp(&local_uuid, &uuid, sizeof(wsrep_uuid_t)) ||
-        local_seqno < seqno || seqno < 0)
+  // Get the locally stored uuid:seqno.
+  if (wsrep_get_SE_checkpoint(local_uuid, local_seqno))
+  {
+    return true;
+  }
+
+  if (memcmp(&local_uuid, &uuid, sizeof(wsrep_uuid_t)) ||
+      local_seqno < seqno)
+  {
+    do_update= true;
+  }
+  else if (local_seqno > seqno)
+  {
+    WSREP_WARN("SST position can't be set in past. Requested: %lld, Current: "
+               " %lld.", (long long)seqno, (long long)local_seqno);
+    /*
+      If we are here because of SET command, simply return true (error) instead of
+      aborting.
+    */
+    if (implicit)
     {
-        wsrep_set_SE_checkpoint(uuid, seqno);
-        local_uuid = uuid;
-        local_seqno = seqno;
+      WSREP_WARN("Can't continue.");
+      unireg_abort(1);
     }
-    else if (local_seqno > seqno)
+    else
     {
-        WSREP_WARN("SST postion is in the past: %lld, current: %lld. "
-                   "Can't continue.",
-                   (long long)seqno, (long long)local_seqno);
-        unireg_abort(1);
+      return true;
     }
+  }
 
 #ifdef GTID_SUPPORT
-    wsrep_init_sidno(uuid);
+  wsrep_init_sidno(uuid);
 #endif /* GTID_SUPPORT */
 
-    if (wsrep)
+  if (wsrep)
+  {
+    int const rcode(seqno < 0 ? seqno : 0);
+    wsrep_gtid_t const state_id= {uuid,
+      (rcode ? WSREP_SEQNO_UNDEFINED : seqno)};
+
+    wsrep_status_t ret= wsrep->sst_received(wsrep, &state_id, state,
+                                            state_len, rcode);
+
+    if (ret != WSREP_OK)
     {
-        int const rcode(seqno < 0 ? seqno : 0);
-        wsrep_gtid_t const state_id = {
-            uuid, (rcode ? WSREP_SEQNO_UNDEFINED : seqno)
-        };
+      return true;
+    }
+  }
 
-        wsrep->sst_received(wsrep, &state_id, state, state_len, rcode);
+  // Now is the good time to update the local state and checkpoint.
+  if (do_update)
+  {
+    if (wsrep_set_SE_checkpoint(uuid, seqno))
+    {
+      return true;
     }
+
+    local_uuid= uuid;
+    local_seqno= seqno;
+  }
+
+  return false;
 }
 
 // Let applier threads to continue
-void wsrep_sst_continue ()
+bool wsrep_sst_continue ()
 {
   if (sst_needed)
   {
     WSREP_INFO("Signalling provider to continue.");
-    wsrep_sst_received (wsrep, local_uuid, local_seqno, NULL, 0);
+    return wsrep_sst_received (wsrep, local_uuid, local_seqno, NULL, 0, true);
   }
+  return false;
 }
 
 struct sst_thread_arg
diff --git a/sql/wsrep_sst.h b/sql/wsrep_sst.h
index 49dd5b39fad..d465f897985 100644
--- a/sql/wsrep_sst.h
+++ b/sql/wsrep_sst.h
@@ -63,7 +63,7 @@ extern void wsrep_sst_grab();
 /*! Init thread waits for SST completion */
 extern bool wsrep_sst_wait();
 /*! Signals wsrep that initialization is complete, writesets can be applied */
-extern void wsrep_sst_continue();
+extern bool wsrep_sst_continue();
 extern void wsrep_sst_auth_free();
 
 extern void wsrep_SE_init_grab();   /*! grab init critical section */
diff --git a/sql/wsrep_var.cc b/sql/wsrep_var.cc
index 9c01e54f48d..1407f715fb4 100644
--- a/sql/wsrep_var.cc
+++ b/sql/wsrep_var.cc
@@ -81,32 +81,68 @@ bool wsrep_sync_wait_update (sys_var* self, THD* thd, enum_var_type var_type)
   return false;
 }
 
-static int wsrep_start_position_verify (const char* start_str)
+
+/*
+  Verify the format of the given UUID:seqno.
+
+  @return
+    true                    Fail
+    false                   Pass
+*/
+static
+bool wsrep_start_position_verify (const char* start_str)
 {
   size_t        start_len;
   wsrep_uuid_t  uuid;
   ssize_t       uuid_len;
 
+  // Check whether it has minimum acceptable length.
   start_len = strlen (start_str);
   if (start_len < 34)
-    return 1;
+    return true;
 
+  /*
+    Parse the input to check whether UUID length is acceptable
+    and seqno has been provided.
+  */
   uuid_len = wsrep_uuid_scan (start_str, start_len, &uuid);
   if (uuid_len < 0 || (start_len - uuid_len) < 2)
-    return 1;
+    return true;
 
-  if (start_str[uuid_len] != ':') // separator should follow UUID
-    return 1;
+  // Separator must follow the UUID.
+  if (start_str[uuid_len] != ':')
+    return true;
 
   char* endptr;
   wsrep_seqno_t const seqno __attribute__((unused)) // to avoid GCC warnings
     (strtoll(&start_str[uuid_len + 1], &endptr, 10));
 
-  if (*endptr == '\0') return 0; // remaining string was seqno
+  // Remaining string was seqno.
+  if (*endptr == '\0') return false;
 
-  return 1;
+  return true;
+}
+
+
+static
+bool wsrep_set_local_position(const char* const value, size_t length,
+                              bool const sst)
+{
+  wsrep_uuid_t uuid;
+  size_t const uuid_len = wsrep_uuid_scan(value, length, &uuid);
+  wsrep_seqno_t const seqno = strtoll(value + uuid_len + 1, NULL, 10);
+
+  if (sst) {
+    return wsrep_sst_received (wsrep, uuid, seqno, NULL, 0, false);
+  } else {
+    // initialization
+    local_uuid = uuid;
+    local_seqno = seqno;
+  }
+  return false;
 }
 
+
 bool wsrep_start_position_check (sys_var *self, THD* thd, set_var* var)
 {
   char start_pos_buf[FN_REFLEN];
@@ -119,52 +155,52 @@ bool wsrep_start_position_check (sys_var *self, THD* thd, set_var* var)
          var->save_result.string_value.length);
   start_pos_buf[var->save_result.string_value.length]= 0;
 
-  if (!wsrep_start_position_verify(start_pos_buf)) return 0;
+  // Verify the format.
+  if (wsrep_start_position_verify(start_pos_buf)) return true;
+
+  /*
+    As part of further verification, we try to update the value and catch
+    errors (if any).
+  */
+  if (wsrep_set_local_position(var->save_result.string_value.str,
+                               var->save_result.string_value.length,
+                               true))
+  {
+    goto err;
+  }
+
+  return false;
 
 err:
   my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), var->var->name.str,
            var->save_result.string_value.str ?
            var->save_result.string_value.str : "NULL");
-  return 1;
-}
-
-static
-void wsrep_set_local_position(const char* const value, bool const sst)
-{
-  size_t const value_len = strlen(value);
-  wsrep_uuid_t uuid;
-  size_t const uuid_len = wsrep_uuid_scan(value, value_len, &uuid);
-  wsrep_seqno_t const seqno = strtoll(value + uuid_len + 1, NULL, 10);
-
-  if (sst) {
-    wsrep_sst_received (wsrep, uuid, seqno, NULL, 0);
-  } else {
-    // initialization
-    local_uuid = uuid;
-    local_seqno = seqno;
-  }
+  return true;
 }
 
 bool wsrep_start_position_update (sys_var *self, THD* thd, enum_var_type type)
 {
-  WSREP_INFO ("wsrep_start_position var submitted: '%s'",
-              wsrep_start_position);
-  // since this value passed wsrep_start_position_check, don't check anything
-  // here
-  wsrep_set_local_position (wsrep_start_position, true);
-  return 0;
+  // Print a confirmation that wsrep_start_position has been updated.
+  WSREP_INFO ("wsrep_start_position set to '%s'", wsrep_start_position);
+  return false;
 }
 
-void wsrep_start_position_init (const char* val)
+bool wsrep_start_position_init (const char* val)
 {
   if (NULL == val || wsrep_start_position_verify (val))
   {
     WSREP_ERROR("Bad initial value for wsrep_start_position: %s", 
                 (val ? val : ""));
-    return;
+    return true;
   }
 
-  wsrep_set_local_position (val, false);
+  if (wsrep_set_local_position (val, strlen(val), false))
+  {
+    WSREP_ERROR("Failed to set initial wsep_start_position: %s", val);
+    return true;
+  }
+
+  return false;
 }
 
 static bool refresh_provider_options()
diff --git a/sql/wsrep_var.h b/sql/wsrep_var.h
index 377b54f2d18..202d6e393e0 100644
--- a/sql/wsrep_var.h
+++ b/sql/wsrep_var.h
@@ -48,7 +48,7 @@ extern bool wsrep_on_update                  UPDATE_ARGS;
 extern bool wsrep_sync_wait_update           UPDATE_ARGS;
 extern bool wsrep_start_position_check       CHECK_ARGS;
 extern bool wsrep_start_position_update      UPDATE_ARGS;
-extern void wsrep_start_position_init        INIT_ARGS;
+extern bool wsrep_start_position_init        INIT_ARGS;
 
 extern bool wsrep_provider_check             CHECK_ARGS;
 extern bool wsrep_provider_update            UPDATE_ARGS;
diff --git a/sql/wsrep_xid.cc b/sql/wsrep_xid.cc
index 133e9cba825..876f791da69 100644
--- a/sql/wsrep_xid.cc
+++ b/sql/wsrep_xid.cc
@@ -89,16 +89,17 @@ static my_bool set_SE_checkpoint(THD* unused, plugin_ref plugin, void* arg)
   return FALSE;
 }
 
-void wsrep_set_SE_checkpoint(XID& xid)
+bool wsrep_set_SE_checkpoint(XID& xid)
 {
-  plugin_foreach(NULL, set_SE_checkpoint, MYSQL_STORAGE_ENGINE_PLUGIN, &xid);
+  return plugin_foreach(NULL, set_SE_checkpoint, MYSQL_STORAGE_ENGINE_PLUGIN,
+                        &xid);
 }
 
-void wsrep_set_SE_checkpoint(const wsrep_uuid_t& uuid, wsrep_seqno_t seqno)
+bool wsrep_set_SE_checkpoint(const wsrep_uuid_t& uuid, wsrep_seqno_t seqno)
 {
   XID xid;
   wsrep_xid_init(&xid, uuid, seqno);
-  wsrep_set_SE_checkpoint(xid);
+  return wsrep_set_SE_checkpoint(xid);
 }
 
 static my_bool get_SE_checkpoint(THD* unused, plugin_ref plugin, void* arg)
@@ -118,12 +119,13 @@ static my_bool get_SE_checkpoint(THD* unused, plugin_ref plugin, void* arg)
   return FALSE;
 }
 
-void wsrep_get_SE_checkpoint(XID& xid)
+bool wsrep_get_SE_checkpoint(XID& xid)
 {
-  plugin_foreach(NULL, get_SE_checkpoint, MYSQL_STORAGE_ENGINE_PLUGIN, &xid);
+  return plugin_foreach(NULL, get_SE_checkpoint, MYSQL_STORAGE_ENGINE_PLUGIN,
+                        &xid);
 }
 
-void wsrep_get_SE_checkpoint(wsrep_uuid_t& uuid, wsrep_seqno_t& seqno)
+bool wsrep_get_SE_checkpoint(wsrep_uuid_t& uuid, wsrep_seqno_t& seqno)
 {
   uuid= WSREP_UUID_UNDEFINED;
   seqno= WSREP_SEQNO_UNDEFINED;
@@ -132,16 +134,24 @@ void wsrep_get_SE_checkpoint(wsrep_uuid_t& uuid, wsrep_seqno_t& seqno)
   memset(&xid, 0, sizeof(xid));
   xid.formatID= -1;
 
-  wsrep_get_SE_checkpoint(xid);
+  if (wsrep_get_SE_checkpoint(xid))
+  {
+    return true;
+  }
 
-  if (xid.formatID == -1) return; // nil XID
+  if (xid.formatID == -1)                       // nil XID
+  {
+    return false;
+  }
 
   if (!wsrep_is_wsrep_xid(&xid))
   {
     WSREP_WARN("Read non-wsrep XID from storage engines.");
-    return;
+    return false;
   }
 
   uuid= *wsrep_xid_uuid(xid);
   seqno= wsrep_xid_seqno(xid);
+
+  return false;
 }
diff --git a/sql/wsrep_xid.h b/sql/wsrep_xid.h
index c3cad0231d7..5b33a904de1 100644
--- a/sql/wsrep_xid.h
+++ b/sql/wsrep_xid.h
@@ -28,9 +28,9 @@ const wsrep_uuid_t* wsrep_xid_uuid(const XID&);
 wsrep_seqno_t wsrep_xid_seqno(const XID&);
 
 //void wsrep_get_SE_checkpoint(XID&);             /* uncomment if needed */
-void wsrep_get_SE_checkpoint(wsrep_uuid_t&, wsrep_seqno_t&);
+bool wsrep_get_SE_checkpoint(wsrep_uuid_t&, wsrep_seqno_t&);
 //void wsrep_set_SE_checkpoint(XID&);             /* uncomment if needed */
-void wsrep_set_SE_checkpoint(const wsrep_uuid_t&, wsrep_seqno_t);
+bool wsrep_set_SE_checkpoint(const wsrep_uuid_t&, wsrep_seqno_t);
 
 #endif /* WITH_WSREP */
 #endif /* WSREP_UTILS_H */