MDEV-26266 Fix error handling around remove_fragments()

Handle the case where client_service::remove_fragments() fails,
for reasons other than bf abort. In this case, we want to make
sure that the transaction state is moved to s_must_abort, so
that we satisfy the sanity check at the end of before_prepare():

```
   assert(state() == s_preparing ||
          (is_xa() && state() == s_replaying) ||
          (ret && (state() == s_must_abort ||
                   state() == s_must_replay ||
                   state() == s_cert_failed ||
                   state() == s_aborted)));
```

src/transaction.cpp

@@ -312,6 +312,12 @@ int wsrep::transaction::before_prepare(
                 ret = client_service_.remove_fragments();
                 if (ret)
                 {
+                    lock.lock();
+                    if (state() == s_executing)
+                    {
+                        state(lock, s_must_abort);
+                    }
+                    lock.unlock();
                     client_state_.override_error(wsrep::e_deadlock_error);
                 }
             }

test/mock_client_state.hpp

@@ -69,6 +69,7 @@ namespace wsrep
             , bf_abort_during_wait_()
             , bf_abort_during_fragment_removal_()
             , error_during_prepare_data_()
+            , error_during_fragment_removal_(false)
             , killed_before_certify_()
             , sync_point_enabled_()
             , sync_point_action_()
@@ -99,10 +100,13 @@ namespace wsrep
                 client_state_->after_rollback();
                 return 1;
             }
-            else
+
+            if (error_during_fragment_removal_)
             {
-                return 0;
+                return 1;
             }
+
+            return 0;
         }
 
         void will_replay() WSREP_OVERRIDE { will_replay_called_ = true; }
@@ -218,6 +222,7 @@ namespace wsrep
         bool bf_abort_during_wait_;
         bool bf_abort_during_fragment_removal_;
         bool error_during_prepare_data_;
+        bool error_during_fragment_removal_;
         bool killed_before_certify_;
         std::string sync_point_enabled_;
         enum sync_point_action

test/transaction_test.cpp

@@ -1388,6 +1388,52 @@ BOOST_FIXTURE_TEST_CASE(transaction_streaming_1pc_bf_abort_during_fragment_remov
                                             wsrep::transaction_id(1));
 }
 
+//
+// Test error during fragment removal.
+//
+
+BOOST_FIXTURE_TEST_CASE(transaction_streaming_1pc_error_during_fragment_removal,
+                        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);
+
+    cc.error_during_fragment_removal_ = true;
+
+    // Run before commit
+    BOOST_REQUIRE(cc.before_commit());
+    BOOST_REQUIRE(tc.state() == wsrep::transaction::s_must_abort);
+
+    cc.error_during_fragment_removal_ = false;
+
+    BOOST_REQUIRE(cc.current_error() == wsrep::e_deadlock_error);
+    BOOST_REQUIRE(tc.certified() == false);
+    BOOST_REQUIRE(tc.ordered() == false);
+
+    // Rollback sequence
+    BOOST_REQUIRE(cc.before_rollback() == 0);
+    BOOST_REQUIRE(tc.state() == wsrep::transaction::s_aborting);
+    BOOST_REQUIRE(cc.after_rollback() == 0);
+    BOOST_REQUIRE(tc.state() == wsrep::transaction::s_aborted);
+
+    // Cleanup after statement
+    cc.after_statement();
+    BOOST_REQUIRE(tc.active() == false);
+    BOOST_REQUIRE(tc.ordered() == false);
+    BOOST_REQUIRE(tc.certified() == false);
+    BOOST_REQUIRE(cc.current_error());
+
+
+    wsrep::high_priority_service* hps(
+        sc.find_streaming_applier(sc.id(), wsrep::transaction_id(1)));
+    BOOST_REQUIRE(hps);
+    hps->rollback(wsrep::ws_handle(), wsrep::ws_meta());
+    hps->after_apply();
+    sc.stop_streaming_applier(sc.id(), wsrep::transaction_id(1));
+    server_service.release_high_priority_service(hps);
+}
+
 //
 // Test streaming rollback
 //