Pass certification keys also for NBO end.

Certification keys are needed for NBO end to resolve dependencies
for the write sets which follow NBO end. Without keys the following
write sets do not detect dependency to NBO event and may start applying
too early.

include/wsrep/client_state.hpp

@@ -788,8 +788,10 @@ namespace wsrep
          * Begin non-blocking operation phase two. The keys argument
          * passed to this call must contain the same keys which were
          * passed to begin_nbo_phase_one().
+         *
+         * @param keys Key array.
          */
-        int begin_nbo_phase_two();
+        int begin_nbo_phase_two(const wsrep::key_array& keys);
 
         /**
          * End non-blocking operation phase two. This call will
@@ -998,6 +1000,7 @@ namespace wsrep
             wsrep::unique_lock<wsrep::mutex>& lock);
         void update_last_written_gtid(const wsrep::gtid&);
         void debug_log_state(const char*) const;
+        void debug_log_keys(const wsrep::key_array& keys) const;
         void state(wsrep::unique_lock<wsrep::mutex>& lock, enum state state);
         void mode(wsrep::unique_lock<wsrep::mutex>& lock, enum mode mode);
 

include/wsrep/key.hpp

@@ -27,6 +27,10 @@
 
 namespace wsrep
 {
+    /** @class key
+     *
+     * Certification key type.
+     */
     class key
     {
     public:
@@ -44,6 +48,14 @@ namespace wsrep
             , key_parts_len_()
         { }
 
+        /**
+         * Append key part to key.
+         *
+         * @param ptr Pointer to key part data. The caller is supposed to take
+         *            care that the pointer remains valid over the lifetime
+         *            if the key object.
+         * @param len Length of the key part data.
+         */
         void append_key_part(const void* ptr, size_t len)
         {
             if (key_parts_len_ == 3)

include/wsrep/transaction.hpp

@@ -251,7 +251,7 @@ namespace wsrep
         void clear_fragments();
         void cleanup();
         void debug_log_state(const char*) const;
-        void debug_log_key_append(const wsrep::key& key);
+        void debug_log_key_append(const wsrep::key& key) const;
 
         wsrep::server_service& server_service_;
         wsrep::client_service& client_service_;

src/client_state.cpp

@@ -512,6 +512,7 @@ int wsrep::client_state::begin_nbo_phase_one(
     std::chrono::time_point<wsrep::clock> wait_until)
 {
     debug_log_state("begin_nbo_phase_one: enter");
+    debug_log_keys(keys);
     wsrep::unique_lock<wsrep::mutex> lock(mutex_);
     assert(state_ == s_exec);
     assert(mode_ == m_local);
@@ -583,9 +584,10 @@ int wsrep::client_state::enter_nbo_mode(const wsrep::ws_meta& ws_meta)
     return 0;
 }
 
-int wsrep::client_state::begin_nbo_phase_two()
+int wsrep::client_state::begin_nbo_phase_two(const wsrep::key_array& keys)
 {
     debug_log_state("begin_nbo_phase_two: enter");
+    debug_log_keys(keys);
     assert(state_ == s_exec);
     assert(mode_ == m_nbo);
     assert(toi_mode_ == m_undefined);
@@ -597,7 +599,7 @@ int wsrep::client_state::begin_nbo_phase_two()
     // Output stored in nbo_meta_ is copied to toi_meta_ for
     // phase two end.
     enum wsrep::provider::status status(
-        provider().enter_toi(id_, wsrep::key_array(),
+        provider().enter_toi(id_, keys,
                              wsrep::const_buffer(), nbo_meta_,
                              wsrep::provider::flag::commit));
     int ret;
@@ -725,6 +727,18 @@ void wsrep::client_state::debug_log_state(const char* context) const
                     << ",nbo: " << nbo_meta_.seqno() << ")");
 }
 
+void wsrep::client_state::debug_log_keys(const wsrep::key_array& keys) const
+{
+    for (const auto& k : keys)
+    {
+        WSREP_LOG_DEBUG(debug_log_level(),
+                        wsrep::log::debug_level_client_state,
+                        "TOI keys: "
+                        << " id: " << id_
+                        << "key: " << k);
+    }
+}
+
 void wsrep::client_state::state(
     wsrep::unique_lock<wsrep::mutex>& lock WSREP_UNUSED,
     enum wsrep::client_state::state state)

src/transaction.cpp

@@ -1790,7 +1790,7 @@ void wsrep::transaction::debug_log_state(
         << "");
 }
 
-void wsrep::transaction::debug_log_key_append(const wsrep::key& key)
+void wsrep::transaction::debug_log_key_append(const wsrep::key& key) const
 {
     WSREP_LOG_DEBUG(client_state_.debug_log_level(),
                     wsrep::log::debug_level_transaction,