Move async task cleanup before transaction destruction

arangodb · goedderz · Jun 3, 2025 · Apr 28, 2025 · May 14, 2025 · May 20, 2025
commit 4cc6436dae3f17873639d8b8a2004cca7c6ad6ad
diff --git a/arangod/Aql/ExecutionEngine.cpp b/arangod/Aql/ExecutionEngine.cpp
@@ -267,14 +267,7 @@ ExecutionEngine::~ExecutionEngine() {
     std::this_thread::sleep_for(10ms);
   }
 
-  // We need to stop prefetch tasks in topological order so
-  // that after stopping tasks on a certain node there is no prefetch
-  // tasks running on any dependent which could start another on the
-  // current block.
-  // The blocks are pushed in a reversed topological order.
-  for (auto it = _blocks.rbegin(); it != _blocks.rend(); ++it) {
-    (*it)->stopAsyncTasks();
-  }
+  stopAsyncTasks();
 
   if (_sharedState) {  // ensure no async task is working anymore
     _sharedState->invalidate();
@@ -971,6 +964,22 @@ ExecutionEngine::rebootTrackers() {
   return _rebootTrackers;
 }
 
+void ExecutionEngine::stopAsyncTasks() {
+  TRI_IF_FAILURE("AsyncPrefetch::blocksDestroyedOutOfOrder") {
+    using namespace std::chrono_literals;
+    std::this_thread::sleep_for(10ms);
+  }
+
+  // We need to stop prefetch tasks in topological order so
+  // that after stopping tasks on a certain node there is no prefetch
+  // tasks running on any dependent which could start another on the
+  // current block.
+  // The blocks are pushed in a reversed topological order.
+  for (auto it = _blocks.rbegin(); it != _blocks.rend(); ++it) {
+    (*it)->stopAsyncTasks();
+  }
+}
+
 std::shared_ptr<SharedQueryState> const& ExecutionEngine::sharedState() const {
   return _sharedState;
 }

diff --git a/arangod/Aql/ExecutionEngine.h b/arangod/Aql/ExecutionEngine.h
@@ -130,6 +130,8 @@ class ExecutionEngine {
 
   std::vector<arangodb::cluster::CallbackGuard>& rebootTrackers();
 
+  void stopAsyncTasks();
+
 #ifdef USE_ENTERPRISE
   static bool parallelizeGraphNode(
       aql::Query& query, ExecutionPlan& plan, aql::GraphNode* graphNode,

diff --git a/arangod/Aql/Query.cpp b/arangod/Aql/Query.cpp
@@ -1852,6 +1852,11 @@ void Query::enterState(QueryExecutionState::ValueType state) {
 /// @brief cleanup plan and engine for current query
 ExecutionState Query::cleanupPlanAndEngine(bool sync) {
   ensureExecutionTime();
+  // Before transaction is destroyed we should wait for all async tasks to
+  // finish so they don't use trx object
+  for (auto const&
4C5D
 snippet : _snippets) {
+    snippet->stopAsyncTasks();
+  }
 
   {
     std::unique_lock<std::mutex> guard{_resultMutex};