refactor: Implement cancellation for Delay() timers. (#524)

This implements CancellationSource / CancellationToken, based on the
similar concepts in C++20 and other languages. It also updates the spots
where we race timers to cancel any timers that aren't finished, so that
they don't pile up internally.

This is mostly Claude-generated code, but I have read through it and
verified it is correct.

There are alternatives to this model for simpler cases, but I like that
this implementation means we could eventually use it for other cases
(like cancelling http requests), and since all the things racing share a
cancellation source, they can all be cancelled together.

This PR does not implement cancellation for HTTP requests yet, since
that is much more complicated, and because they are less likely to pile
up during normal operation.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Introduces new cross-thread cancellation primitives and changes timer
scheduling/cancellation behavior, which can affect concurrency,
ordering, and potential deadlocks/leaks if incorrect. Scope is limited
to async utilities and FDv2 polling timeouts/delays, with added tests to
validate the new behavior.
> 
> **Overview**
> Adds a new
`CancellationSource`/`CancellationToken`/`CancellationCallback`
primitive (C++20 `stop_*`-style) to support coordinated cancellation
with RAII callback registration and destructor synchronization.
> 
> Extends `async::Delay` to accept an optional `CancellationToken` and
cancels the underlying ASIO timer when triggered, while also hardening
`Continuation` construction to avoid self-recursion. Updates FDv2
polling (`polling_synchronizer`) to create a shared cancellation source
for each `WhenAny` race and cancel the non-winning delay/timeout timers
to prevent timer buildup, and adds comprehensive unit tests for
cancellation and timer interaction.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
a187254. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: beekld

libs/internal/include/launchdarkly/async/cancellation.hpp

@@ -0,0 +1,242 @@
+#pragma once
+
+#include <launchdarkly/async/promise.hpp>
+
+#include <condition_variable>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <thread>
+
+// This file implements a cancellation primitive modelled after C++20's
+// std::stop_source / std::stop_token / std::stop_callback design: a source
+// owns the ability to trigger cancellation, lightweight tokens (derived from
+// the source) can be freely passed around, and CancellationCallback provides
+// RAII registration of a callback tied to a token.
+
+namespace launchdarkly::async {
+
+// CancellationState is the shared state between a CancellationSource and all
+// CancellationTokens and CancellationCallbacks derived from it. This is an
+// internal class; use CancellationSource, CancellationToken, and
+// CancellationCallback instead.
+class CancellationState {
+   public:
+    using CallbackId = std::size_t;
+
+    // Sentinel returned by Register() when the state was already cancelled;
+    // the callback is invoked immediately in that case. Deregister() is a
+    // no-op for this value.
+    static constexpr CallbackId kAlreadyCancelled = 0;
+
+    CancellationState() = default;
+    ~CancellationState() = default;
+    CancellationState(CancellationState const&) = delete;
+    CancellationState& operator=(CancellationState const&) = delete;
+    CancellationState(CancellationState&&) = delete;
+    CancellationState& operator=(CancellationState&&) = delete;
+
+    // Registers a callback and returns its ID. If Cancel() has already been
+    // called, invokes cb immediately (outside the lock) and returns
+    // kAlreadyCancelled.
+    CallbackId Register(Continuation<void()> cb) {
+        std::unique_lock lock(mutex_);
+        if (cancelled_) {
+            lock.unlock();
+            cb();
+            return kAlreadyCancelled;
+        }
+        CallbackId id = next_id_++;
+        callbacks_.emplace(id, std::move(cb));
+        return id;
+    }
+
+    // Deregisters the callback with the given ID. If the callback is currently
+    // executing on another thread, blocks until execution completes. This
+    // mirrors the synchronization guarantee of C++20's stop_callback
+    // destructor: after Deregister returns, the callback is guaranteed to have
+    // either never run or fully completed. No-op if id is kAlreadyCancelled or
+    // the callback has already run.
+    void Deregister(CallbackId id) {
+        if (id == kAlreadyCancelled) {
+            return;
+        }
+
+        std::unique_lock lock(mutex_);
+
+        // The callback is still pending: remove it before it can run.
+        if (callbacks_.erase(id)) {
+            return;
+        }
+
+        // The callback has already run, or was never registered.
+        if (executing_id_ != id) {
+            return;
+        }
+
+        // The callback is executing on this thread (re-entrant call from
+        // within the callback itself): return without waiting to avoid
+        // deadlock.
+        if (executing_thread_ == std::this_thread::get_id()) {
+            return;
+        }
+
+        // The callback is executing on another thread. Wait for it to
+        // finish. executing_id_ is set while the state lock is held — before
+        // unlocking for invocation — so there is no window where the callback
+        // is running but executing_id_ is not yet set.
+        executing_done_.wait(lock, [this, id] { return executing_id_ != id; });
+    }
+
+    // Invokes all registered callbacks in registration order, then clears the
+    // pending list. Callbacks are executed one at a time with the lock
+    // released during each invocation to prevent deadlocks. No-op if called
+    // more than once.
+    void Cancel() {
+        std::unique_lock lock(mutex_);
+        if (cancelled_) {
+            return;
+        }
+        cancelled_ = true;
+
+        while (!callbacks_.empty()) {
+            // Extract the next entry while still holding the lock, then set
+            // executing_id_ before releasing. This ensures Deregister can
+            // never observe a window where the callback is running but
+            // executing_id_ is unset.
+            auto node = callbacks_.extract(callbacks_.begin());
+            executing_id_ = node.key();
+            executing_thread_ = std::this_thread::get_id();
+
+            lock.unlock();
+            node.mapped()();
+            lock.lock();
+
+            executing_id_ = kAlreadyCancelled;
+            lock.unlock();
+            executing_done_.notify_all();
+            lock.lock();
+        }
+    }
+
+    // Returns true if Cancel() has been called.
+    bool IsCancelled() const {
+        std::lock_guard lock(mutex_);
+        return cancelled_;
+    }
+
+   private:
+    mutable std::mutex mutex_;
+    bool cancelled_ = false;
+    CallbackId next_id_ = 1;  // Real IDs start at 1; 0 is kAlreadyCancelled.
+    std::map<CallbackId, Continuation<void()>> callbacks_;
+
+    // Tracks which callback (if any) is currently being invoked by Cancel(),
+    // and on which thread, to support the blocking destructor in Deregister.
+    CallbackId executing_id_ = kAlreadyCancelled;
+    std::thread::id executing_thread_;
+    std::condition_variable executing_done_;
+};
+
+class CancellationToken;
+
+// CancellationSource is the write end of a cancellation pair: call Cancel()
+// to signal all operations holding tokens derived from this source.
+//
+// CancellationSource is copyable; copies share the same underlying
+// CancellationState, matching the behaviour of C++20's stop_source.
+class CancellationSource {
+   public:
+    CancellationSource() : state_(std::make_shared<CancellationState>()) {}
+
+    ~CancellationSource() = default;
+    CancellationSource(CancellationSource const&) = default;
+    CancellationSource& operator=(CancellationSource const&) = default;
+    CancellationSource(CancellationSource&&) = default;
+    CancellationSource& operator=(CancellationSource&&) = default;
+
+    // Invokes all registered callbacks in registration order. No-op if called
+    // more than once.
+    void Cancel() { state_->Cancel(); }
+
+    // Returns true if Cancel() has been called.
+    bool IsCancelled() const { return state_->IsCancelled(); }
+
+    // Returns a token referring to this source's cancellation state. The
+    // token may be freely copied and passed to any number of
+    // CancellationCallbacks.
+    CancellationToken GetToken() const;
+
+   private:
+    std::shared_ptr<CancellationState> state_;
+};
+
+// CancellationToken is the read end of a cancellation pair. Tokens are
+// obtained from CancellationSource::GetToken() and passed to
+// CancellationCallback constructors to register callbacks.
+//
+// A default-constructed token has no associated state: any
+// CancellationCallback constructed from it is never invoked.
+//
+// CancellationToken is cheap to copy; all copies share the same underlying
+// CancellationState.
+class CancellationToken {
+   public:
+    CancellationToken() = default;
+
+    explicit CancellationToken(std::shared_ptr<CancellationState> state)
+        : state_(std::move(state)) {}
+
+    // Returns true if the associated source has been cancelled, or false if
+    // there is no associated source.
+    bool IsCancelled() const { return state_ && state_->IsCancelled(); }
+
+   private:
+    std::shared_ptr<CancellationState> state_;
+
+    friend class CancellationCallback;
+};
+
+inline CancellationToken CancellationSource::GetToken() const {
+    return CancellationToken(state_);
+}
+
+// CancellationCallback registers a callback to be invoked when the associated
+// CancellationSource is cancelled. The callback is invoked on whichever thread
+// calls CancellationSource::Cancel().
+//
+// The design follows C++20's std::stop_callback:
+//
+//   - Constructing a CancellationCallback registers the callback. If the
+//     source was already cancelled, the callback is invoked immediately in the
+//     constructor.
+//   - Destroying a CancellationCallback deregisters the callback. If the
+//     callback is currently executing on another thread, the destructor blocks
+//     until execution completes, preventing use-after-free of anything captured
+//     by the callback.
+//   - CancellationCallback is non-copyable and non-movable, matching C++20's
+//     stop_callback.
+class CancellationCallback {
+   public:
+    CancellationCallback(CancellationToken token, Continuation<void()> cb)
+        : state_(token.state_),
+          id_(state_ ? state_->Register(std::move(cb))
+                     : CancellationState::kAlreadyCancelled) {}
+
+    ~CancellationCallback() {
+        if (state_) {
+            state_->Deregister(id_);
+        }
+    }
+
+    CancellationCallback(CancellationCallback const&) = delete;
+    CancellationCallback& operator=(CancellationCallback const&) = delete;
+    CancellationCallback(CancellationCallback&&) = delete;
+    CancellationCallback& operator=(CancellationCallback&&) = delete;
+
+   private:
+    std::shared_ptr<CancellationState> state_;
+    CancellationState::CallbackId id_;
+};
+
+}  // namespace launchdarkly::async

libs/internal/include/launchdarkly/async/promise.hpp

@@ -54,7 +54,9 @@ class Continuation<R(Args...)> {
     // function pointer, or other callable; it need not be copy-constructible.
     // F&& is a forwarding reference: accepts any callable by move or copy,
     // then moves it into Impl<F> so Continuation itself owns the callable.
-    template <typename F>
+    template <typename F,
+              typename = std::enable_if_t<
+                  !std::is_same_v<std::decay_t<F>, Continuation>>>
     Continuation(F&& f)
         : impl_(std::make_unique<Impl<std::decay_t<F>>>(std::forward<F>(f))) {}
     Continuation(Continuation&&) = default;

libs/internal/include/launchdarkly/async/timer.hpp

@@ -1,9 +1,11 @@
 #pragma once
 
+#include <launchdarkly/async/cancellation.hpp>
 #include <launchdarkly/async/promise.hpp>
 
 #include <boost/asio/any_io_executor.hpp>
 #include <boost/asio/error.hpp>
+#include <boost/asio/post.hpp>
 #include <boost/asio/steady_timer.hpp>
 #include <boost/system/error_code.hpp>
 
@@ -15,17 +17,55 @@ namespace launchdarkly::async {
 // Returns a Future<bool> that resolves once the given duration elapses.
 // The future resolves with true if the timer fired normally, or false if
 // the timer was cancelled before it expired.
+//
+// If a CancellationToken is provided, cancelling the associated
+// CancellationSource cancels the timer, resolving the future with false.
 template <typename Rep, typename Period>
 Future<bool> Delay(boost::asio::any_io_executor executor,
-                   std::chrono::duration<Rep, Period> duration) {
+                   std::chrono::duration<Rep, Period> duration,
+                   CancellationToken token = {}) {
     auto timer = std::make_shared<boost::asio::steady_timer>(executor);
     timer->expires_after(duration);
     Promise<bool> promise;
     auto future = promise.GetFuture();
-    timer->async_wait([p = std::move(promise),
-                       timer](boost::system::error_code code) mutable {
+
+    // This code is tricky because there are a few constraints that conflict.
+    // 1. We need to make sure timer->cancel isn't called _before_
+    //    timer->async_wait, or else it'll just be ignored.
+    // 2. The cancellation_callback has to be created _before_
+    //    timer->async_wait, because it has to be captured by async_wait's
+    //    handler, because it is an RAII type, and once it is destroyed, it
+    //    deregisters itself. It has to stay alive as long as the timer needs to
+    //    be cancellable.
+
+    Promise<std::monostate> timer_started_promise;
+    Future<std::monostate> timer_started_future =
+        timer_started_promise.GetFuture();
+
+    auto cancel_timer = [timer, executor,
+                         timer_started_future =
+                             std::move(timer_started_future)]() mutable {
+        timer_started_future.Then(
+            [timer](auto const&) -> std::monostate {
+                timer->cancel();
+                return {};
+            },
+            [executor](Continuation<void()> f) {
+                boost::asio::post(executor, std::move(f));
+            });
+    };
+
+    auto cancellation_callback =
+        std::make_shared<CancellationCallback>(std::move(token), cancel_timer);
+
+    timer->async_wait([p = std::move(promise), timer, cancellation_callback](
+                          boost::system::error_code code) mutable {
+        cancellation_callback.reset();
         p.Resolve(code != boost::asio::error::operation_aborted);
     });
+
+    timer_started_promise.Resolve({});
+
     return future;
 }