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].

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,
-                         const wsrep_uuid_t&       uuid,
-                         wsrep_seqno_t       const seqno,
-                         const void*         const state,
-                         size_t              const state_len)
-{
-    wsrep_get_SE_checkpoint(local_uuid, local_seqno);
+/*
+  If wsrep provider is loaded, inform that the new state snapshot
+  has been received. Also update the local checkpoint.
 
-    if (memcmp(&local_uuid, &uuid, sizeof(wsrep_uuid_t)) ||
-        local_seqno < seqno || seqno < 0)
+  @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,
+                         const wsrep_seqno_t       seqno,
+                         const void*         const state,
+                         const size_t              state_len,
+                         const bool                implicit)
+{
+  /*
+    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;
+
+  // 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)
-    {
-        int const rcode(seqno < 0 ? seqno : 0);
-        wsrep_gtid_t const state_id = {
-            uuid, (rcode ? WSREP_SEQNO_UNDEFINED : seqno)
-        };
+  if (wsrep)
+  {
+    int const rcode(seqno < 0 ? seqno : 0);
+    wsrep_gtid_t const state_id= {uuid,
+      (rcode ? WSREP_SEQNO_UNDEFINED : seqno)};
 
-        wsrep->sst_received(wsrep, &state_id, state, state_len, rcode);
+    wsrep_status_t ret= wsrep->sst_received(wsrep, &state_id, state,
+                                            state_len, rcode);
+
+    if (ret != WSREP_OK)
+    {
+      return true;
     }
+  }
+
+  // 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