Abbondanzo · Oct 12, 2026
diff --git a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp
@@ -50,18 +50,15 @@ TaskDispatchThread::TaskDispatchThread(
 #else
   thread_ = std::thread(&TaskDispatchThread::loop, this);
 #endif
+  threadId_ = thread_.get_id();
 }
 
 TaskDispatchThread::~TaskDispatchThread() noexcept {
   quit();
-
-  if (thread_.joinable()) {
-    thread_.join();
-  }
 }
 
 bool TaskDispatchThread::isOnThread() noexcept {
-  return std::this_thread::get_id() == thread_.get_id();
+  return !loopStopped_ && std::this_thread::get_id() == threadId_;
 }
 
 bool TaskDispatchThread::isRunning() noexcept {
@@ -71,12 +68,13 @@ bool TaskDispatchThread::isRunning() noexcept {
 void TaskDispatchThread::runAsync(
     TaskFn&& task,
     std::chrono::milliseconds delayMs) noexcept {
+  auto ownedTask = std::make_unique<TaskFn>(std::move(task));
   std::lock_guard<std::mutex> guard(mutex_);
   if (!running_) {
     return;
   }
   auto dispatchTime = std::chrono::system_clock::now() + delayMs;
-  queue_.emplace(dispatchTime, std::move(task));
+  queue_.emplace(dispatchTime, std::move(ownedTask));
   loopCv_.notify_one();
 }
 
@@ -97,61 +95,62 @@ void TaskDispatchThread::runSync(TaskFn&& task) noexcept {
   promise->get_future().wait();
 }
 
-void TaskDispatchThread::quit() noexcept {
+void TaskDispatchThread::cancelPendingTasks() noexcept {
+  std::priority_queue<Task> discarded;
   {
-    /*
-      We should hold lock even when changing atomic variable to correctly
-      publish the modificatino to the waiting thread
-      from: https://en.cppreference.com/w/cpp/thread/condition_variable.html
-    */
-    std::unique_lock<std::mutex> lock(mutex_);
-    if (!running_.exchange(false)) {
-      return;
-    }
+    std::lock_guard<std::mutex> lock(mutex_);
+    queue_.swap(discarded);
   }
+  loopCv_.notify_one();
+}
 
+void TaskDispatchThread::quit() noexcept {
+  const bool onThread = isOnThread();
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    running_ = false;
+  }
   loopCv_.notify_one();
-  if (!isOnThread()) {
-    loopStoppedPromise_.get_future().wait();
+  if (!onThread) {
+    {
+      std::lock_guard<std::mutex> lock(joinMutex_);
+      if (thread_.joinable()) {
+        thread_.join();
+      }
+    }
   }
+  cancelPendingTasks();
 }
 
 void TaskDispatchThread::loop() noexcept {
   if (!threadName_.empty()) {
     folly::setThreadName(threadName_);
   }
+  std::unique_lock<std::mutex> lock(mutex_);
   while (running_) {
-    std::unique_lock<std::mutex> lock(mutex_);
     loopCv_.wait(lock, [&]() { return !running_ || !queue_.empty(); });
-
-    while (!queue_.empty()) {
-      auto task = queue_.top();
-      auto now = std::chrono::system_clock::now();
-
-      if (running_) {
-        if (task.dispatchTime > now) {
-          // Wait until the scheduled task time, if delayed
-          loopCv_.wait_until(lock, task.dispatchTime);
-          continue;
-        }
-      } else {
-        // Shutting down, skip all the remaining tasks
-        queue_ = {};
-        break;
-      }
-
-      // We should check whether the task thread is still running at this
-      // point (which may not anymore be the case since the previous check)
+    if (!running_) {
+      break;
+    }
+    const auto dispatchTime = queue_.top().dispatchTime;
+    if (dispatchTime > std::chrono::system_clock::now()) {
+      loopCv_.wait_until(lock, dispatchTime, [this, dispatchTime]() {
+        return !running_ || queue_.empty() ||
+            queue_.top().dispatchTime != dispatchTime;
+      });
+      continue;
+    }
+    {
+      auto fn = std::move(queue_.top().fn);
+      queue_.pop();
+      lock.unlock();
       if (running_) {
-        lock.unlock();
-        task.fn();
-        lock.lock();
+        (*fn)();
       }
-      queue_.pop();
     }
+    lock.lock();
   }
-
-  loopStoppedPromise_.set_value();
+  loopStopped_ = true;
 }
 
 } //  namespace facebook::react
diff --git a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h
@@ -7,10 +7,12 @@
 
 #pragma once
 
+#include <atomic>
 #include <chrono>
 #include <condition_variable>
 #include <functional>
 #include <future>
+#include <memory>
 #include <mutex>
 #include <queue>
 #include <thread>
@@ -42,15 +44,18 @@ class TaskDispatchThread {
   /** Add task to the queue and wait until it has completed. */
   void runSync(TaskFn &&task) noexcept;
 
+  void cancelPendingTasks() noexcept;
+
   /** Shut down and clean up the thread. */
   void quit() noexcept;
 
  protected:
   struct Task {
     TimePoint dispatchTime;
-    TaskFn fn;
+    // Moving the callback must preserve the heap's deadline ordering key.
+    mutable std::unique_ptr<TaskFn> fn;
 
-    Task(TimePoint dispatchTime, TaskFn &&fn) : dispatchTime(dispatchTime), fn(std::move(fn)) {}
+    Task(TimePoint dispatchTime, std::unique_ptr<TaskFn> &&fn) : dispatchTime(dispatchTime), fn(std::move(fn)) {}
 
     bool operator<(const Task &other) const
     {
@@ -67,8 +72,10 @@ class TaskDispatchThread {
   std::atomic<bool> running_{true};
   std::string threadName_;
   std::thread thread_;
+  std::thread::id threadId_;
+  std::mutex joinMutex_;
 
-  std::promise<void> loopStoppedPromise_;
+  std::atomic<bool> loopStopped_{false};
 };
 
 } // namespace facebook::react
diff --git a/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp b/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp
@@ -10,6 +10,7 @@
 #include <atomic>
 #include <chrono>
 #include <future>
+#include <memory>
 #include <thread>
 
 namespace facebook::react {
@@ -216,4 +217,137 @@ TEST_F(TaskDispatchThreadTest, RunSyncDoesNotBlockWhenQuitCalledConcurrently) {
     localDispatcher.reset();
   }
 }
+
+TEST_F(TaskDispatchThreadTest, ExternalQuitJoinsEvenAfterWorkerRequestedQuit) {
+  std::promise<void> entered;
+  std::promise<void> release;
+  auto released = release.get_future();
+  dispatcher->runAsync([&]() {
+    dispatcher->quit();
+    entered.set_value();
+    released.wait();
+  });
+  auto ready = entered.get_future().wait_for(std::chrono::seconds(5));
+  if (ready != std::future_status::ready) {
+    release.set_value();
+    FAIL() << "Worker did not reach the shutdown barrier";
+  }
+
+  std::promise<void> joined;
+  auto joinedFuture = joined.get_future();
+  std::thread owner([&]() {
+    dispatcher->quit();
+    joined.set_value();
+  });
+  EXPECT_EQ(
+      joinedFuture.wait_for(std::chrono::seconds(1)),
+      std::future_status::timeout);
+  release.set_value();
+  owner.join();
+  EXPECT_EQ(
+      joinedFuture.wait_for(std::chrono::seconds(0)),
+      std::future_status::ready);
+  EXPECT_FALSE(dispatcher->isRunning());
+}
+
+TEST_F(TaskDispatchThreadTest, ConcurrentExternalQuitWaitsForInFlightCallback) {
+  std::promise<void> entered;
+  std::promise<void> release;
+  auto released = release.get_future();
+  std::atomic<bool> finished{false};
+  dispatcher->runAsync([&]() {
+    entered.set_value();
+    released.wait();
+    finished = true;
+  });
+  auto ready = entered.get_future().wait_for(std::chrono::seconds(5));
+  if (ready != std::future_status::ready) {
+    release.set_value();
+    FAIL() << "Worker did not reach the callback barrier";
+  }
+  std::promise<void> firstJoined;
+  std::promise<void> secondJoined;
+  auto firstFuture = firstJoined.get_future();
+  auto secondFuture = secondJoined.get_future();
+  std::thread first([&]() {
+    dispatcher->quit();
+    EXPECT_TRUE(finished);
+    firstJoined.set_value();
+  });
+  std::thread second([&]() {
+    dispatcher->quit();
+    EXPECT_TRUE(finished);
+    secondJoined.set_value();
+  });
+  EXPECT_EQ(
+      firstFuture.wait_for(std::chrono::seconds(1)),
+      std::future_status::timeout);
+  EXPECT_EQ(
+      secondFuture.wait_for(std::chrono::seconds(1)),
+      std::future_status::timeout);
+  release.set_value();
+  first.join();
+  second.join();
+  EXPECT_TRUE(finished);
+}
+
+TEST(
+    TaskDispatchThreadShutdownTest,
+    DiscardedCapturesAreReleasedAfterJoinWithoutQueueLock) {
+  class Dispatcher final : public TaskDispatchThread {
+   public:
+    bool queueMutexAvailable() {
+      std::unique_lock<std::mutex> lock(mutex_, std::try_to_lock);
+      return lock.owns_lock();
+    }
+  } dispatcher;
+  bool released = false;
+  auto ownerThread = std::this_thread::get_id();
+  auto token = std::shared_ptr<void>(nullptr, [&](void*) {
+    EXPECT_TRUE(dispatcher.queueMutexAvailable());
+    EXPECT_EQ(std::this_thread::get_id(), ownerThread);
+    released = true;
+  });
+  dispatcher.runAsync([token]() {}, std::chrono::hours(1));
+  token.reset();
+
+  dispatcher.quit();
+
+  EXPECT_TRUE(released);
+}
+
+TEST_F(TaskDispatchThreadTest, DispatchDoesNotCopyCallbackOnWorker) {
+  struct State {
+    TaskDispatchThread* dispatcher{nullptr};
+    std::atomic<int> copiesOnWorker{0};
+    std::promise<void> ran;
+  };
+  struct CopyCountingTask {
+    std::shared_ptr<State> state;
+
+    explicit CopyCountingTask(std::shared_ptr<State> state)
+        : state(std::move(state)) {}
+
+    CopyCountingTask(const CopyCountingTask& other) noexcept
+        : state(other.state) {
+      if (state->dispatcher->isOnThread()) {
+        ++state->copiesOnWorker;
+      }
+    }
+
+    CopyCountingTask& operator=(const CopyCountingTask&) = default;
+
+    void operator()() {
+      state->ran.set_value();
+    }
+  };
+  auto state = std::make_shared<State>();
+  state->dispatcher = dispatcher.get();
+  auto ran = state->ran.get_future();
+  dispatcher->runAsync(CopyCountingTask(state));
+  ASSERT_EQ(ran.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  dispatcher->quit();
+  EXPECT_EQ(state->copiesOnWorker.load(), 0);
+}
+
 } // namespace facebook::react