Preserve async cancellation on task completion

cpp/async/handle_base.hpp

@@ -237,6 +237,9 @@ class handle_base
     inline void set_value(T&& value)
     {
         std::unique_lock lock(data_->lock_);
+        if (is_cancelled(data_)) {
+            return;
+        }
         ASSERT(!has_result(data_));
         data_->result_ = Result(std::forward<T>(value));
         on_result_updated(std::move(lock), data_);
@@ -247,6 +250,9 @@ class handle_base
     inline void set_value(T&& value)
     {
         std::unique_lock lock(data_->lock_);
+        if (is_cancelled(data_)) {
+            return;
+        }
         ASSERT(!has_result(data_));
         data_->result_ = Result(std::forward<T>(value));
         on_result_updated(std::move(lock), data_);
@@ -257,6 +263,9 @@ class handle_base
     requires(std::is_void_v<Result>)
     {
         std::unique_lock lock(data_->lock_);
+        if (is_cancelled(data_)) {
+            return;
+        }
         ASSERT(!has_result(data_));
         data_->result_->set_fulfilled();
         on_result_updated(std::move(lock), data_);
@@ -265,6 +274,9 @@ class handle_base
     inline void set_exception(std::exception_ptr ex)
     {
         std::unique_lock lock(data_->lock_);
+        if (is_cancelled(data_)) {
+            return;
+        }
         ASSERT(!has_result(data_) || !data_->result_->succeeded());
         data_->result_ = std::move(ex);
         on_result_updated(std::move(lock), data_);

cpp/async/promise.hpp

@@ -237,7 +237,8 @@ class promise
      * or re-raises the exception in case of failure.
      *
      * @pre The promise should have no callback registered. This is a mutually exclusive interface with set_callback.
-     * @warning The future will be forever locked if the promise will be cancelled.
+     * @warning Successful cancellation clears the callback and makes the future ready with
+     * std::future_errc::broken_promise, rather than delivering a result value.
      */
     [[nodiscard]] inline std::future<T> get_future()
     {

cpp/async/tests/README.md

@@ -0,0 +1,22 @@
+# Cancellation result regression
+
+`cancelled_result_test.cpp` uses the real `bg_queue_promise::call`, `handle_base`,
+`result`, and spin-lock headers. A worker is held inside the producer after its
+initial cancellation check; the main thread cancels the handle and then lets the
+producer return a value, return void, or throw. Cancellation must remain terminal,
+and neither the original callback nor a later subscriber may run. Normal
+completion still delivers its value.
+
+For standalone header validation, compile with C++20, pthreads, Abseil and
+`-DASYNC_STANDALONE_QUEUE_TEST`, including `cpp/` and a directory containing the
+build's `config.hpp`. This flag supplies only queue-linkage stubs: the queue ID is
+unsubmitted, callbacks explicitly use the inline/null queue, and any actual
+scheduler execution terminates the test. It does not replace the producer,
+result-state or locking implementation. Do not define that flag when linking to
+the complete async queue implementation.
+
+The focused standalone check does not exercise a real queue scheduler or the
+full Deep Lake extension. A separate check exercises the real `promise::get_future`:
+successful cancellation destroys the callback's captured `std::promise`, making
+the future ready with `std::future_errc::broken_promise`. The previous warning that
+the future remained locked was stale; the regression preserves the actual behavior.

cpp/async/tests/cancelled_result_test.cpp

@@ -0,0 +1,93 @@
+// Exercises the real producer/handle/result headers with cancellation during work.
+// Standalone builds isolate only queue linkage; no scheduler is exercised.
+#include <async/impl/bg_queue_promise.hpp>
+#include <async/promise.hpp>
+
+#include <cassert>
+#include <future>
+#include <iostream>
+#include <stdexcept>
+#include <thread>
+
+#ifdef ASYNC_STANDALONE_QUEUE_TEST
+namespace async {
+queue::~queue() = default;
+main_queue::main_queue() : queue(nullptr) {}
+main_queue& main() noexcept { static main_queue value; return value; }
+queue::id_type::~id_type() noexcept = default;
+void queue::id_type::remove() const noexcept { assert(!static_cast<bool>(*this)); }
+void queue::id_type::set_priority(int) const { assert(!static_cast<bool>(*this)); }
+bool queue::is_this_thread_worker() const noexcept { std::terminate(); }
+void queue::submit(base::function<void()>&&, int, id_type*) { std::terminate(); }
+} // namespace async
+#endif
+
+void cancelled_public_promise_drops_late_value()
+{
+    async::impl::bg_queue_promise<int> handle;
+    async::promise<int> promise(handle);
+    int callbacks = 0;
+    promise.set_callback(async::callback_type<int>([&](async::result<int>&&) { ++callbacks; }, nullptr));
+    assert(promise.cancel());
+    handle.set_value(42);
+    if (!promise.is_cancelled() || promise.is_ready() || callbacks != 0) {
+        throw std::runtime_error("late value resurrected a promise cancelled through its public API");
+    }
+    promise.set_callback(async::callback_type<int>([&](async::result<int>&&) { ++callbacks; }, nullptr));
+    assert(callbacks == 0);
+}
+
+template <typename T, typename Producer>
+void cancelled_running_producer(Producer produce)
+{
+    async::impl::bg_queue_promise<T> handle;
+    int callbacks = 0;
+    handle.set_callback(async::callback_type<T>([&](async::result<T>&&) { ++callbacks; }, nullptr));
+    std::promise<void> entered;
+    std::promise<void> release;
+    auto proceed = release.get_future();
+    std::thread worker([&] {
+        handle.call([&]() -> T {
+            entered.set_value();
+            proceed.wait();
+            return produce();
+        });
+    });
+    entered.get_future().wait(); // call() has passed its initial cancellation check.
+    assert(handle.cancel());
+    release.set_value();
+    worker.join();
+    if (!handle.is_cancelled() || handle.is_ready() || callbacks != 0) {
+        throw std::runtime_error("late producer result resurrected a cancelled handle");
+    }
+    handle.set_callback(async::callback_type<T>([&](async::result<T>&&) { ++callbacks; }, nullptr));
+    assert(callbacks == 0); // Cancellation remains terminal for later subscribers too.
+}
+
+int main()
+{
+    cancelled_public_promise_drops_late_value();
+    cancelled_running_producer<int>([] { return 42; });
+    cancelled_running_producer<void>([] {});
+    cancelled_running_producer<int>([]() -> int { throw std::runtime_error("producer failed"); });
+    cancelled_running_producer<void>([] { throw std::runtime_error("producer failed"); });
+    async::handle_base<int> normal;
+    int delivered = 0;
+    normal.set_callback(async::callback_type<int>([&](async::result<int>&& value) { delivered = std::move(value).get(); }, nullptr));
+    normal.set_value(7);
+    assert(delivered == 7);
+    async::impl::bg_queue_promise<int> waiting;
+    async::promise<int> promise(waiting);
+    auto future = promise.get_future();
+    assert(promise.cancel());
+    waiting.set_value(99);
+    assert(promise.is_cancelled());
+    assert(future.wait_for(std::chrono::milliseconds(0)) == std::future_status::ready);
+    try {
+        (void)future.get();
+        assert(false);
+    } catch (const std::future_error& error) {
+        assert(error.code() == std::make_error_code(std::future_errc::broken_promise));
+    }
+    std::cout << "public cancel then set_value, four cancelled producers, future cancellation and normal completion passed\n";
+}