Fix crash close_orphaned_sr_transactions() BF abort

Some implementations of provider `bf_abort()` require `victim_ctx`
to point to application side victim context. However, when
total order BF abort was done from `close_orphaned_sr_transactions()`,
the application side context was not available.

To fix this, added an interface method `call_in_operation_context()`
which allows invoking a function object with a reference to
application side operation context passed in as a parameter.
This method is used in `close_orphaned_sr_transactions()`
to call `client_state::total_order_bf_abort()` with appropriate
victim operation context.

dbsim/db_client_service.hpp

@@ -92,6 +92,14 @@ namespace db
             return false;
         }
 
+        void call_in_operation_context(
+            const std::function<void(wsrep::operation_context&)>& fn)
+            const override
+        {
+            wsrep::null_operation_context op;
+            fn(op);
+        }
+
         void debug_sync(const char*) override { }
         void debug_crash(const char*) override { }
     private:

include/wsrep/client_service.hpp

@@ -28,10 +28,13 @@
 #define WSREP_CLIENT_SERVICE_HPP
 
 #include "buffer.hpp"
+#include "operation_context.hpp"
 #include "provider.hpp"
 #include "mutex.hpp"
 #include "lock.hpp"
 
+#include <functional>
+
 namespace wsrep
 {
     class client_service
@@ -212,6 +215,15 @@ namespace wsrep
          */
         virtual bool is_xa_rollback() = 0;
 
+        /**
+         * Perform a function call in operation context associated
+         * with the client service. The implementation must call @p fn
+         * with operation_context set appropriately.
+         */
+        virtual void call_in_operation_context(
+            const std::function<void(wsrep::operation_context&)>& fn) const
+            = 0;
+
         //
         // Debug interface
         //

src/server_state.cpp

@@ -1487,6 +1487,7 @@ void wsrep::server_state::close_orphaned_sr_transactions(
         {
             wsrep::client_id client_id(i->first);
             wsrep::transaction_id transaction_id(i->second->transaction().id());
+            auto& client_state = *i->second;
             // It is safe to unlock the server state temporarily here.
             // The processing happens inside view handler which is
             // protected by the provider commit ordering critical
@@ -1497,7 +1498,15 @@ void wsrep::server_state::close_orphaned_sr_transactions(
             // remains unlocked, so it should not be accessed after
             // the bf abort call.
             lock.unlock();
-            i->second->total_order_bf_abort(current_view_.view_seqno());
+            client_state.client_service().call_in_operation_context(
+                [&client_state, this](wsrep::operation_context& op_ctx)
+                {
+                    wsrep::unique_lock<wsrep::mutex> lock{
+                        client_state.mutex()
+                    };
+                    client_state.total_order_bf_abort(
+                        lock, current_view_.view_seqno(), op_ctx);
+                });
             lock.lock();
             streaming_clients_map::const_iterator found_i;
             while ((found_i = streaming_clients_.find(client_id)) !=

test/mock_client_state.hpp

@@ -186,6 +186,14 @@ namespace wsrep
             return false;
         }
 
+        void call_in_operation_context(
+            const std::function<void(wsrep::operation_context&)>& fn) const
+            WSREP_OVERRIDE
+        {
+            wsrep::null_operation_context op;
+            fn(op);
+        }
+
         void debug_sync(const char* sync_point) WSREP_OVERRIDE
         {
             if (sync_point_enabled_ == sync_point)