Enable DBMS initialization from applying codepath

In general the provider might not guarantee that after joining
to replication group, the first applied event is view event.

Made init_initialized_ atomic and check it also from applying
codepath. If the DBMS is not yet initialized when applying the
first event, change state to initializing and wait for DBMS
initialization completion.

Notes:
- Allow connected -> initializing state change.
- Handle bootstrap_view in mock_connect to make server state synced
  for tests.

include/wsrep/server_state.hpp

@@ -92,10 +92,13 @@
 #include "compiler.hpp"
 #include "xid.hpp"
 
+#include <atomic>
 #include <deque>
 #include <vector>
 #include <string>
 #include <map>
+#include <string>
+#include <vector>
 
 /**
  * Magic string to tell provider to engage into trivial (empty)
@@ -686,7 +689,7 @@ namespace wsrep
         mutable std::vector<int> state_waiters_;
         bool bootstrap_;
         const wsrep::gtid initial_position_;
-        bool init_initialized_;
+        std::atomic<bool> init_initialized_;
         bool init_synced_;
         wsrep::gtid sst_gtid_;
         size_t desync_count_;

src/server_state.cpp

@@ -1098,6 +1098,26 @@ int wsrep::server_state::on_apply(
     const wsrep::ws_meta& ws_meta,
     const wsrep::const_buffer& data)
 {
+    if (not init_initialized_.load())
+    {
+        wsrep::unique_lock<wsrep::mutex> lock(mutex_);
+        if (state(lock) == s_connected)
+        {
+            state(lock, s_initializing);
+            lock.unlock();
+            server_service_.debug_sync("on_view_wait_initialized");
+            lock.lock();
+        }
+        while (not init_initialized_.load())
+        {
+            wait_until_state(lock, s_initialized);
+        }
+        if (state(lock) == s_initialized)
+        {
+            state(lock, s_joined);
+        }
+    }
+
     if (is_toi(ws_meta.flags()))
     {
         return apply_toi(provider(), high_priority_service,
@@ -1330,7 +1350,7 @@ void wsrep::server_state::state(
             {  0,   1,   0,    1,    0,   0,   0,   0,   0}, /* dis */
             {  1,   0,   1,    0,    0,   0,   0,   0,   1}, /* ing */
             {  1,   0,   0,    1,    0,   1,   0,   0,   1}, /* ized */
-            {  1,   0,   0,    1,    1,   0,   0,   1,   1}, /* cted */
+            {  1,   1,   0,    1,    1,   0,   0,   1,   1}, /* cted */
             {  1,   1,   0,    0,    0,   1,   0,   0,   1}, /* jer */
             {  1,   0,   0,    1,    0,   0,   1,   1,   1}, /* jed */
             {  1,   0,   0,    1,    0,   1,   0,   0,   1}, /* dor */

test/mock_server_state.hpp

@@ -278,6 +278,14 @@ namespace wsrep
                                            1,
                                            members);
                 server_state::on_connect(bootstrap_view);
+                server_state::initialized();
+                wsrep::mock_client cs(*this, wsrep::client_id(0),
+                                      wsrep::client_state::m_high_priority);
+                wsrep::mock_high_priority_service hps(*this, &cs, false);
+                server_state::on_view(bootstrap_view, &hps);
+                BOOST_REQUIRE(state() == wsrep::server_state::s_joined);
+                server_state::on_sync();
+                BOOST_REQUIRE(state() == wsrep::server_state::s_synced);
             }
             else
             {

test/server_context_test.cpp

@@ -485,6 +485,22 @@ BOOST_FIXTURE_TEST_CASE(
     BOOST_REQUIRE(ss.state() == wsrep::server_state::s_disconnected);
 }
 
+BOOST_FIXTURE_TEST_CASE(
+    server_state_sst_first_init_on_apply,
+    sst_first_server_fixture)
+{
+    connect_in_view(second_view);
+    BOOST_REQUIRE(ss.state() == wsrep::server_state::s_connected);
+    sst_received_action();
+    char buf[1] = { 1 };
+    BOOST_REQUIRE(ss.on_apply(hps, ws_handle, ws_meta,
+                              wsrep::const_buffer(buf, 1)) == 0);
+    const wsrep::transaction& txc(cc.transaction());
+    BOOST_REQUIRE(txc.state() == wsrep::transaction::s_committed);
+    BOOST_REQUIRE(ss.state() == wsrep::server_state::s_joined);
+
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 //                     Test cases for init first                             //
 ///////////////////////////////////////////////////////////////////////////////