schedulers: run_loop and new_thread re-arm the timer wait when an earlier schedulable is queued

src/rpp/rpp/schedulers/new_thread.hpp

@@ -87,7 +87,10 @@ namespace rpp::schedulers
                     {
                         if (const auto now = worker_strategy::now(); now < state->queue.top()->get_timepoint())
                         {
-                            state->cv.wait_for(lock, state->queue.top()->get_timepoint() - now, [&] { return state->queue.top()->is_disposed() || worker_strategy::now() >= state->queue.top()->get_timepoint(); });
+                            // Sleep until the current top is due, and wake as soon as the top changes
+                            // (an earlier schedulable was queued), so the loop recomputes the wait.
+                            const auto* const top = state->queue.top().get();
+                            state->cv.wait_until(lock, top->get_timepoint(), [&] { return state->queue.is_empty() || state->queue.top().get() != top || state->queue.top()->is_disposed() || worker_strategy::now() >= state->queue.top()->get_timepoint(); });
                             continue;
                         }
                     }

src/rpp/rpp/schedulers/run_loop.hpp

@@ -66,7 +66,10 @@ namespace rpp::schedulers
                     if (!wait)
                         break;
 
-                    m_cv.wait_for(lock, m_queue.top()->get_timepoint() - now, [&]() { return is_disposed() || (!m_queue.is_empty() && m_queue.top()->get_timepoint() <= worker_strategy::now()); });
+                    // Sleep until the current top is due, and wake as soon as the top changes
+                    // (an earlier schedulable was queued), so the loop recomputes the wait.
+                    const auto* const top = m_queue.top().get();
+                    m_cv.wait_until(lock, top->get_timepoint(), [&]() { return is_disposed() || m_queue.is_empty() || m_queue.top().get() != top || m_queue.top()->is_disposed() || m_queue.top()->get_timepoint() <= worker_strategy::now(); });
                 }
                 return {};
             }

src/tests/rpp/test_scheduler.cpp

@@ -508,7 +508,7 @@
             executions.push_back(rpp::schedulers::clock_type::now());
             if (++call_count <= 1)
                 return rpp::schedulers::delay_from_this_timepoint{diff};
             return {};
         },
                          obs.value());
 
@@ -526,7 +526,7 @@
             executions.push_back(rpp::schedulers::clock_type::now());
             if (++call_count <= 1)
                 return rpp::schedulers::delay_to{executions[0] + diff};
             return {};
         },
                          obs.value());
 
@@ -1068,3 +1068,72 @@
 
     CHECK(f.get());
 }
+
+TEST_CASE("timer queued while the worker waits for a later one runs at its own time")
+{
+    // The worker is already waiting for a schedulable due in `later`. A second one, due
+    // in `earlier` (much sooner), is queued from another thread. It has to run after
+    // about `earlier`, not when the wait for the first one ends.
+    constexpr auto later   = std::chrono::seconds{3};
+    constexpr auto earlier = std::chrono::milliseconds{50};
+    constexpr auto bound   = std::chrono::seconds{1};
+
+    auto       d    = rpp::composite_disposable_wrapper::make();
+    auto       obs  = mock_observer_strategy<int>{}.get_observer(d).as_dynamic();
+    const auto noop = [](const auto&) {
+        return rpp::schedulers::optional_delay_from_now{};
+    };
+
+    SUBCASE("run_loop")
+    {
+        auto scheduler = rpp::schedulers::run_loop{};
+        auto worker    = scheduler.create_worker();
+        worker.schedule(later, noop, obs);
+
+        std::atomic_bool                      earlier_ran{};
+        std::chrono::steady_clock::time_point queued_at{};
+        std::chrono::steady_clock::time_point ran_at{};
+
+        auto other = std::thread{[&] {
+            std::this_thread::sleep_for(std::chrono::milliseconds{100}); // dispatch() below waits for `later` by now
+            queued_at = std::chrono::steady_clock::now();
+            worker.schedule(
+                earlier,
+                [&](const auto&) {
+                    ran_at = std::chrono::steady_clock::now();
+                    earlier_ran.store(true);
+                    return rpp::schedulers::optional_delay_from_now{};
+                },
+                obs);
+        }};
+
+        while (!earlier_ran.load())
+            scheduler.dispatch();
+        other.join();
+
+        CHECK(ran_at - queued_at < bound);
+    }
+
+    SUBCASE("new_thread")
+    {
+        auto worker = rpp::schedulers::new_thread::create_worker();
+        worker.schedule(later, noop, obs);
+        std::this_thread::sleep_for(std::chrono::milliseconds{100}); // the worker thread waits for `later` by now
+
+        std::promise<std::chrono::steady_clock::time_point> ran_at{};
+        const auto                                          queued_at = std::chrono::steady_clock::now();
+        worker.schedule(
+            earlier,
+            [&ran_at](const auto&) {
+                ran_at.set_value(std::chrono::steady_clock::now());
+                return rpp::schedulers::optional_delay_from_now{};
+            },
+            obs);
+
+        auto f = ran_at.get_future();
+        REQUIRE(f.wait_for(later * 2) == std::future_status::ready);
+        CHECK(f.get() - queued_at < bound);
+    }
+
+    d.dispose();
+}