Merge branch 'main' into f/hehehe

Author: maxtropets

src/tasks/sub_task_queue.h

@@ -10,7 +10,8 @@ namespace ccf::tasks
 {
   // Helper type for OrderedTasks, containing a list of sub-tasks to be
   // performed in-order. Modifiers return bools indicating whether the caller
-  // is responsible for scheduling a future flush of this queue.
+  // should schedule processing of this queue *now* (eg, enqueue the owning
+  // runner), based on the current state.
   template <typename T>
   class SubTaskQueue
   {
@@ -21,10 +22,19 @@ namespace ccf::tasks
     std::atomic<bool> paused;
 
   public:
+    // Enqueue a new sub-task.
+    //
+    // Returns true iff this call made the queue non-empty, the queue was not
+    // already active, and the queue is not paused. Callers should interpret
+    // a true return as "schedule processing of this queue now" (eg, enqueue
+    // the parent runner). While paused, this function intentionally never
+    // requests scheduling (it will always return false), even if the queue
+    // transitions from empty to non-empty. This avoids double-enqueue when
+    // the queue is resumed elsewhere.
     bool push(T&& t)
     {
       std::lock_guard<std::mutex> lock(pending_mutex);
-      const bool ret = pending.empty() && !active.load();
+      const bool ret = pending.empty() && !active.load() && !paused.load();
       pending.emplace_back(std::forward<T>(t));
       return ret;
     }

src/tasks/test/ordered_tasks.cpp

@@ -199,3 +199,61 @@ TEST_CASE("OrderedTasks" * doctest::test_suite("ordered_tasks"))
     }
   }
 }
+
+TEST_CASE(
+  "Concurrent pause + add_action does not double-enqueue" *
+  doctest::test_suite("ordered_tasks"))
+{
+  ccf::tasks::JobBoard job_board;
+  auto tasks = ccf::tasks::OrderedTasks::create(job_board);
+
+  std::vector<size_t> execution_order;
+
+  // Step 1: Add an action that pauses itself mid-execution
+  ccf::tasks::Resumable resumable;
+
+  tasks->add_action(ccf::tasks::make_basic_action([&]() {
+    execution_order.push_back(1);
+
+    // Pause the task (simulating respond_on_commit)
+    resumable = ccf::tasks::pause_current_task();
+  }));
+
+  // Step 2: Execute the first action - it will pause
+  {
+    auto task = job_board.get_task();
+    REQUIRE(task != nullptr);
+    task->do_task();
+  }
+
+  // Confirm results of action - board should be empty, and we hold a resumable
+  // token to restore the paused queue
+  REQUIRE(job_board.get_task() == nullptr);
+  REQUIRE(resumable != nullptr);
+
+  // Step 3: Simulate concurrent operations — add_action + resume_task
+  tasks->add_action(
+    ccf::tasks::make_basic_action([&]() { execution_order.push_back(2); }));
+
+  ccf::tasks::resume_task(std::move(resumable));
+
+  // Step 4: Count how many times the task was enqueued
+  size_t enqueue_count = 0;
+  while (true)
+  {
+    auto task = job_board.get_task();
+    if (task == nullptr)
+    {
+      break;
+    }
+    enqueue_count++;
+    task->do_task();
+  }
+
+  // Confirm that despite 2 potentially-queuing concurrent operations, only one
+  // actual enqueue occurred
+  REQUIRE(enqueue_count == 1);
+
+  // Verify the second action (added in step 3) actually executed
+  REQUIRE(execution_order == std::vector<size_t>{1, 2});
+}