Change unit tests to mimic application behavior on replaying

Allocate separate client state in mock_client_service::replay()
for replaying step.

Added two new test cases for streaming replication replay after
BF abort.

test/mock_client_state.cpp

@@ -35,14 +35,42 @@ int wsrep::mock_client_service::bf_rollback()
     return ret;
 }
 
+struct replayer_context
+{
+    wsrep::mock_client_state state;
+    wsrep::mock_client_service service;
+    replayer_context(wsrep::server_state& server_state,
+                     const wsrep::transaction& transaction,
+                     const wsrep::client_id& id)
+        : state{server_state, service, id, wsrep::client_state::m_high_priority}
+        , service{&state}
+    {
+        state.open(id);
+        state.before_command();
+        state.clone_transaction_for_replay(transaction);
+    }
+
+    ~replayer_context() {
+        state.after_applying();
+        state.after_command_before_result();
+        state.after_command_after_result();
+        state.close();
+    }
+};
+
 enum wsrep::provider::status
 wsrep::mock_client_service::replay()
 {
-    wsrep::mock_high_priority_service hps(client_state_->server_state(),
+    /* Mimic application and allocate separate client state for replaying. */
-                                          client_state_, true);
+    wsrep::client_id replayer_id{ 1001 };
+    replayer_context replayer(client_state_->server_state(),
+                              client_state_->transaction(), replayer_id);
+    wsrep::mock_high_priority_service hps{ client_state_->server_state(),
+                                           &replayer.state, true };
+
     enum wsrep::provider::status ret(
         client_state_->provider().replay(
-            client_state_->transaction().ws_handle(),
+            replayer.state.transaction().ws_handle(),
             &hps));
     ++replays_;
     return ret;

test/mock_provider.hpp

@@ -201,7 +201,6 @@ namespace wsrep
             wsrep::mock_client_state& cc(
                 static_cast<wsrep::mock_client_state&>(
                     high_priority_service.client_state()));
-            wsrep::high_priority_context high_priority_context(cc);
             const wsrep::transaction& tc(cc.transaction());
             wsrep::ws_meta ws_meta;
             if (replay_result_ == wsrep::provider::success)

test/transaction_test.cpp

@@ -1539,6 +1539,55 @@ BOOST_FIXTURE_TEST_CASE(
     BOOST_REQUIRE(sc.provider().commit_fragments() == 1);
 }
 
+//
+// Test a streaming transaction which gets BF aborted inside provider before
+// certification result is known. Replaying will be successful
+//
+BOOST_FIXTURE_TEST_CASE(
+    transaction_row_streaming_bf_before_cert_result_replay_success,
+    streaming_client_fixture_row)
+{
+    BOOST_REQUIRE(cc.start_transaction(wsrep::transaction_id(1)) == 0);
+    BOOST_REQUIRE(cc.after_row() == 0);
+    BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
+
+    sc.provider().certify_result_ = wsrep::provider::error_bf_abort;
+    sc.provider().replay_result_ = wsrep::provider::success;
+
+    BOOST_REQUIRE(cc.before_commit());
+    BOOST_REQUIRE(tc.state() == wsrep::transaction::s_must_replay);
+    BOOST_REQUIRE(cc.will_replay_called() == true);
+    BOOST_REQUIRE(cc.after_statement() == 0);
+    BOOST_REQUIRE(tc.state() == wsrep::transaction::s_committed);
+    BOOST_REQUIRE(cc.current_error() == wsrep::e_success);
+}
+
+//
+// Test a streaming transaction which gets BF aborted inside provider before
+// certification result is known. Replaying will fail because of
+// certification failure.
+//
+BOOST_FIXTURE_TEST_CASE(
+    transaction_row_streaming_bf_before_cert_result_replay_cert_fail,
+    streaming_client_fixture_row)
+{
+    BOOST_REQUIRE(cc.start_transaction(wsrep::transaction_id(1)) == 0);
+    BOOST_REQUIRE(cc.after_row() == 0);
+    BOOST_REQUIRE(tc.streaming_context().fragments_certified() == 1);
+
+    sc.provider().certify_result_ = wsrep::provider::error_bf_abort;
+    sc.provider().replay_result_ = wsrep::provider::error_certification_failed;
+
+    BOOST_REQUIRE(cc.before_commit());
+    BOOST_REQUIRE(tc.state() == wsrep::transaction::s_must_replay);
+    BOOST_REQUIRE(cc.will_replay_called() == true);
+    BOOST_REQUIRE(cc.after_statement() );
+    BOOST_REQUIRE(tc.state() == wsrep::transaction::s_aborted);
+    BOOST_REQUIRE(cc.current_error() == wsrep::e_deadlock_error);
+    BOOST_REQUIRE(tc.active() == false);
+}
+
+
 BOOST_FIXTURE_TEST_CASE(transaction_byte_streaming_1pc_commit,
                         streaming_client_fixture_byte)
 {