diff --git a/CHANGELOG.md b/CHANGELOG.md
index 60ad4998..fcb2ba76 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -67,6 +67,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   Directly constructed public send awaiters remain owning, and queue,
   cancellation, close, and dequeue-versus-destruction behavior is unchanged
   (#1047).
+- **Compact cancellation callback dispatch**: Cancellation registrations now
+  synchronize selection, invocation, and teardown through a native-width
+  atomic phase instead of embedding a mutex and condition variable in every
+  callback node. Immediate cancellation, LIFO dispatch, same-dispatch
+  suppression, cross-source reentrant teardown, exception aggregation, and
+  exactly-once payload destruction retain their existing contracts (#1053).
 - **Allocation-free ready semaphore acquires**: Non-cancellable acquires now
   defer shared wake-state allocation until the ready permit check fails.
   Entering `await_suspend` allocates before taking the semaphore queue lock,
diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt
index 4e96d8a5..123d3893 100644
--- a/examples/CMakeLists.txt
+++ b/examples/CMakeLists.txt
@@ -50,6 +50,10 @@ add_executable(microbench microbench.cpp)
 target_link_libraries(microbench PRIVATE elio)
 target_link_options(microbench PRIVATE ${STATIC_LINK_FLAGS})
 
+add_executable(cancel_callback_benchmark cancel_callback_benchmark.cpp)
+target_link_libraries(cancel_callback_benchmark PRIVATE elio)
+target_link_options(cancel_callback_benchmark PRIVATE ${STATIC_LINK_FLAGS})
+
 add_executable(scheduler_service_benchmark scheduler_service_benchmark.cpp)
 target_link_libraries(scheduler_service_benchmark PRIVATE elio)
 target_link_options(scheduler_service_benchmark PRIVATE ${STATIC_LINK_FLAGS})
diff --git a/examples/cancel_callback_benchmark.cpp b/examples/cancel_callback_benchmark.cpp
new file mode 100644
index 00000000..ede17ba7
--- /dev/null
+++ b/examples/cancel_callback_benchmark.cpp
@@ -0,0 +1,294 @@
+#include <elio/coro/cancel_token.hpp>
+
+#include <algorithm>
+#include <atomic>
+#include <charconv>
+#include <chrono>
+#include <cstddef>
+#include <cstdlib>
+#include <iomanip>
+#include <iostream>
+#include <new>
+#include <numeric>
+#include <string>
+#include <string_view>
+#include <system_error>
+#include <thread>
+#include <vector>
+
+#if defined(__GNUC__) || defined(__clang__)
+#define ELIO_BENCH_NOINLINE __attribute__((noinline))
+#else
+#define ELIO_BENCH_NOINLINE
+#endif
+
+namespace allocation_probe {
+thread_local bool enabled = false;
+thread_local std::size_t allocations = 0;
+thread_local std::size_t requested_bytes = 0;
+}
+
+ELIO_BENCH_NOINLINE void* allocate_for_probe(std::size_t size) {
+    if (void* ptr = std::malloc(size == 0 ? 1 : size)) {
+        if (allocation_probe::enabled) {
+            ++allocation_probe::allocations;
+            allocation_probe::requested_bytes += size;
+        }
+        return ptr;
+    }
+    throw std::bad_alloc();
+}
+
+ELIO_BENCH_NOINLINE void deallocate_for_probe(void* ptr) noexcept {
+    std::free(ptr);
+}
+
+ELIO_BENCH_NOINLINE void* operator new(std::size_t size) {
+    return allocate_for_probe(size);
+}
+
+ELIO_BENCH_NOINLINE void* operator new[](std::size_t size) {
+    return allocate_for_probe(size);
+}
+
+ELIO_BENCH_NOINLINE void operator delete(void* ptr) noexcept {
+    deallocate_for_probe(ptr);
+}
+
+ELIO_BENCH_NOINLINE void operator delete[](void* ptr) noexcept {
+    deallocate_for_probe(ptr);
+}
+
+ELIO_BENCH_NOINLINE void operator delete(void* ptr, std::size_t) noexcept {
+    deallocate_for_probe(ptr);
+}
+
+ELIO_BENCH_NOINLINE void operator delete[](void* ptr, std::size_t) noexcept {
+    deallocate_for_probe(ptr);
+}
+
+#undef ELIO_BENCH_NOINLINE
+
+namespace {
+
+using clock_type = std::chrono::steady_clock;
+using elio::coro::cancel_registration;
+using elio::coro::cancel_source;
+
+template<typename F>
+double time_ns(F&& operation) {
+    const auto start = clock_type::now();
+    operation();
+    const auto end = clock_type::now();
+    return std::chrono::duration<double, std::nano>(end - start).count();
+}
+
+template<typename F>
+std::vector<double> collect_samples(std::size_t count, F&& operation) {
+    std::vector<double> samples;
+    samples.reserve(count);
+    for (std::size_t i = 0; i < count; ++i) {
+        samples.push_back(operation());
+    }
+    return samples;
+}
+
+double percentile(const std::vector<double>& sorted, double fraction) {
+    const auto index = static_cast<std::size_t>(
+        fraction * static_cast<double>(sorted.size() - 1));
+    return sorted[index];
+}
+
+void report(std::string_view label, std::vector<double> samples) {
+    std::sort(samples.begin(), samples.end());
+    const double mean = std::accumulate(samples.begin(), samples.end(), 0.0) /
+                        static_cast<double>(samples.size());
+    std::cout << std::left << std::setw(42) << label << std::right
+              << " mean=" << std::setw(10) << mean
+              << " p50=" << std::setw(10) << percentile(samples, 0.50)
+              << " p95=" << std::setw(10) << percentile(samples, 0.95)
+              << " p99=" << std::setw(10) << percentile(samples, 0.99)
+              << " ns\n";
+}
+
+std::vector<cancel_registration> register_callbacks(cancel_source& source,
+                                                     std::size_t count) {
+    std::vector<cancel_registration> registrations;
+    registrations.reserve(count);
+    auto token = source.get_token();
+    for (std::size_t i = 0; i < count; ++i) {
+        registrations.push_back(token.on_cancel([] {}));
+    }
+    return registrations;
+}
+
+void benchmark_allocation(std::size_t iterations) {
+    cancel_source source;
+    auto token = source.get_token();
+    std::size_t allocations = 0;
+    std::size_t bytes = 0;
+
+    for (std::size_t i = 0; i < iterations; ++i) {
+        allocation_probe::allocations = 0;
+        allocation_probe::requested_bytes = 0;
+        allocation_probe::enabled = true;
+        auto registration = token.on_cancel([] {});
+        allocation_probe::enabled = false;
+        allocations += allocation_probe::allocations;
+        bytes += allocation_probe::requested_bytes;
+        registration.unregister();
+    }
+
+    std::cout << "callback_node sizeof="
+              << sizeof(elio::coro::detail::callback_node)
+              << " task_parent_callback_node sizeof="
+              << sizeof(elio::coro::detail::task_parent_callback_node) << '\n'
+              << "SBO registration requested allocations="
+              << (static_cast<double>(allocations) / iterations)
+              << " requested bytes="
+              << (static_cast<double>(bytes) / iterations) << "\n\n";
+}
+
+void benchmark_register_unregister(std::size_t iterations) {
+    cancel_source source;
+    auto token = source.get_token();
+    report("register + unregister (one callback)",
+           collect_samples(iterations, [&] {
+               return time_ns([&] {
+                   auto registration = token.on_cancel([] {});
+                   registration.unregister();
+               });
+           }));
+}
+
+void benchmark_unlink(std::size_t iterations, std::size_t list_size,
+                      bool oldest) {
+    report(std::string(oldest ? "unlink oldest / " : "unlink newest / ") +
+               std::to_string(list_size),
+           collect_samples(iterations, [&] {
+               cancel_source source;
+               auto registrations = register_callbacks(source, list_size);
+               auto& target = oldest ? registrations.front()
+                                     : registrations.back();
+               return time_ns([&] { target.unregister(); });
+           }));
+}
+
+void benchmark_dispatch(std::size_t iterations, std::size_t list_size) {
+    report("cancel dispatch / " + std::to_string(list_size),
+           collect_samples(iterations, [&] {
+               cancel_source source;
+               auto registrations = register_callbacks(source, list_size);
+               return time_ns([&] { source.cancel(); });
+           }));
+}
+
+void benchmark_immediate(std::size_t iterations) {
+    std::atomic<std::size_t> invocations{0};
+    report("already-cancelled registration",
+           collect_samples(iterations, [&] {
+               cancel_source source;
+               source.cancel();
+               auto token = source.get_token();
+               return time_ns([&] {
+                   auto registration = token.on_cancel([&] {
+                       invocations.fetch_add(1, std::memory_order_relaxed);
+                   });
+               });
+           }));
+    if (invocations.load(std::memory_order_relaxed) != iterations) {
+        std::abort();
+    }
+}
+
+void benchmark_concurrent_unregister(std::size_t iterations,
+                                     std::size_t list_size) {
+    std::atomic<std::size_t> invocations{0};
+    std::size_t unregister_wins = 0;
+    std::size_t dispatch_wins = 0;
+    report("cancel with concurrent unregister / " +
+               std::to_string(list_size),
+           collect_samples(iterations, [&] {
+               cancel_source source;
+               std::vector<cancel_registration> registrations;
+               registrations.reserve(list_size);
+               auto token = source.get_token();
+               for (std::size_t i = 0; i < list_size; ++i) {
+                   registrations.push_back(token.on_cancel([&] {
+                       invocations.fetch_add(1, std::memory_order_relaxed);
+                   }));
+               }
+
+               const auto invocations_before =
+                   invocations.load(std::memory_order_relaxed);
+               std::atomic<bool> ready{false};
+               std::atomic<bool> start{false};
+               std::thread unregisterer([&] {
+                   ready.store(true, std::memory_order_release);
+                   ready.notify_one();
+                   start.wait(false, std::memory_order_acquire);
+                   for (std::size_t i = 0; i < list_size / 2; ++i) {
+                       registrations[i].unregister();
+                   }
+               });
+
+               ready.wait(false, std::memory_order_acquire);
+               const auto elapsed = time_ns([&] {
+                   start.store(true, std::memory_order_release);
+                   start.notify_one();
+                   source.cancel();
+               });
+               unregisterer.join();
+               const auto iteration_invocations =
+                   invocations.load(std::memory_order_relaxed) -
+                   invocations_before;
+               unregister_wins += list_size - iteration_invocations;
+               dispatch_wins += iteration_invocations;
+               return elapsed;
+           }));
+    std::cout << "  outcomes removed=" << unregister_wins
+              << " invoked=" << dispatch_wins << "\n";
+}
+
+}  // namespace
+
+int main(int argc, char** argv) {
+    std::size_t iterations = 2000;
+    if (argc > 2) {
+        std::cerr << "Usage: " << argv[0]
+                  << " [--smoke|positive-sample-count]\n";
+        return 2;
+    }
+    if (argc == 2) {
+        const std::string_view argument(argv[1]);
+        if (argument == "--smoke") {
+            iterations = 20;
+        } else {
+            const auto* begin = argument.data();
+            const auto* end = begin + argument.size();
+            const auto result =
+                std::from_chars(begin, end, iterations);
+            if (result.ec != std::errc{} || result.ptr != end ||
+                iterations == 0) {
+                std::cerr << "Usage: " << argv[0]
+                          << " [--smoke|positive-sample-count]\n";
+                return 2;
+            }
+        }
+    }
+
+    std::cout << std::fixed << std::setprecision(2)
+              << "samples=" << iterations << '\n';
+    benchmark_allocation(iterations);
+    benchmark_register_unregister(iterations);
+    for (const std::size_t size : {8, 32, 256}) {
+        benchmark_unlink(iterations, size, false);
+        benchmark_unlink(iterations, size, true);
+    }
+    for (const std::size_t size : {1, 8, 32, 256}) {
+        benchmark_dispatch(iterations, size);
+    }
+    benchmark_immediate(iterations);
+    benchmark_concurrent_unregister(
+        std::max<std::size_t>(iterations / 4, 100), 32);
+}
diff --git a/include/elio/coro/cancel_token.hpp b/include/elio/coro/cancel_token.hpp
index 2ac6b724..1c21a210 100644
--- a/include/elio/coro/cancel_token.hpp
+++ b/include/elio/coro/cancel_token.hpp
@@ -1,7 +1,6 @@
 #pragma once
 
 #include <atomic>
-#include <condition_variable>
 #include <coroutine>
 #include <cstddef>
 #include <cstdint>
@@ -10,7 +9,6 @@
 #include <mutex>
 #include <new>
 #include <stdexcept>
-#include <thread>
 #include <type_traits>
 #include <utility>
 
@@ -55,6 +53,28 @@ class callback_dispatch_scope final {
     const void* previous_;
 };
 
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+inline std::atomic<bool> pause_cancel_callback_after_claim_for_test{false};
+inline std::atomic<bool> cancel_callback_paused_after_claim_for_test{false};
+inline std::atomic<bool> pause_cancel_callback_after_invoking_for_test{false};
+inline std::atomic<bool> cancel_callback_paused_after_invoking_for_test{false};
+inline std::atomic<std::size_t> cancel_callback_waiters_for_test{0};
+inline std::atomic<std::size_t> cancel_callback_node_allocations_for_test{0};
+
+inline void pause_cancel_callback_dispatch_for_test(
+    std::atomic<bool>& pause, std::atomic<bool>& paused) noexcept {
+    if (!pause.load(std::memory_order_acquire)) return;
+
+    paused.store(true, std::memory_order_release);
+    paused.notify_all();
+    while (pause.load(std::memory_order_acquire)) {
+        pause.wait(true, std::memory_order_acquire);
+    }
+    paused.store(false, std::memory_order_release);
+    paused.notify_all();
+}
+#endif
+
 /// Type-erased callback node for the cancel_state intrusive list.
 ///
 /// Each registration owns exactly one heap-allocated callback_node. The
@@ -63,7 +83,10 @@ class callback_dispatch_scope final {
 /// single secondary heap allocation. This eliminates the vector growth and
 /// the per-callable std::function allocation of the previous design.
 struct callback_node {
-    enum class phase : uint8_t {
+    // Keep the phase native-width: libstdc++ implements byte-sized atomic
+    // waits through a shared hashed waiter pool instead of the direct futex
+    // path used for a 32-bit atomic.
+    enum class phase : uint32_t {
         registered,
         claimed,
         invoking,
@@ -122,19 +145,31 @@ struct callback_node {
     }
 
     void dispatch() {
-        const void* active_dispatcher;
-        {
-            std::lock_guard<std::mutex> lock(dispatch_mutex);
-            if (dispatch_phase == phase::registered) {
-                invoking_thread = std::this_thread::get_id();
-                dispatcher_identity = this;
-            } else if (dispatch_phase != phase::claimed) {
+        const void* active_dispatcher = this;
+        auto expected = phase::registered;
+        if (!dispatch_phase.compare_exchange_strong(
+                expected, phase::invoking, std::memory_order_acq_rel,
+                std::memory_order_acquire)) {
+            if (expected != phase::claimed) return;
+
+            expected = phase::claimed;
+            if (!dispatch_phase.compare_exchange_strong(
+                    expected, phase::invoking, std::memory_order_acq_rel,
+                    std::memory_order_acquire)) {
                 return;
             }
+            // claim_for_dispatch() published this identity before its release
+            // transition to claimed. The successful acquire above makes it
+            // visible for the whole callback dispatch.
             active_dispatcher = dispatcher_identity;
-            dispatch_phase = phase::invoking;
         }
 
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        pause_cancel_callback_dispatch_for_test(
+            pause_cancel_callback_after_invoking_for_test,
+            cancel_callback_paused_after_invoking_for_test);
+#endif
+
         std::exception_ptr exception;
         {
             callback_dispatch_scope dispatch_scope(active_dispatcher);
@@ -146,53 +181,60 @@ struct callback_node {
             destroy_payload();
         }
 
-        {
-            std::lock_guard<std::mutex> lock(dispatch_mutex);
-            invoking_thread = {};
-            dispatcher_identity = nullptr;
-            dispatch_phase = phase::completed;
-        }
-        dispatch_cv.notify_all();
+        dispatch_phase.store(phase::completed, std::memory_order_release);
+        dispatch_phase.notify_all();
 
         if (exception) {
             std::rethrow_exception(exception);
         }
     }
 
-    void claim_for_dispatch(std::thread::id dispatch_thread,
-                            const void* dispatcher) noexcept {
-        std::lock_guard<std::mutex> lock(dispatch_mutex);
-        if (dispatch_phase != phase::registered) return;
-        invoking_thread = dispatch_thread;
+    void claim_for_dispatch(const void* dispatcher) noexcept {
+        // The cancel-state mutex excludes registration teardown until after
+        // the node is detached and this selection is published.
+        if (dispatch_phase.load(std::memory_order_relaxed) !=
+            phase::registered) {
+            return;
+        }
         dispatcher_identity = dispatcher;
-        dispatch_phase = phase::claimed;
+        dispatch_phase.store(phase::claimed, std::memory_order_release);
     }
 
     void unregister_and_wait() noexcept {
-        std::unique_lock<std::mutex> lock(dispatch_mutex);
-        if (dispatch_phase == phase::registered) {
-            dispatch_phase = phase::unregistered;
-            lock.unlock();
-            destroy_payload();
-            return;
-        }
+        auto observed = dispatch_phase.load(std::memory_order_acquire);
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        bool announced_waiter = false;
+#endif
+        while (observed == phase::registered || observed == phase::claimed ||
+               observed == phase::invoking) {
+            if (observed == phase::registered) {
+                if (dispatch_phase.compare_exchange_weak(
+                        observed, phase::unregistered,
+                        std::memory_order_acq_rel,
+                        std::memory_order_acquire)) {
+                    destroy_payload();
+                    return;
+                }
+                continue;
+            }
 
-        if (dispatch_phase == phase::claimed &&
-            dispatcher_identity == current_callback_dispatcher) {
-            // A callback may unregister a later callback selected by the same
-            // synchronous cancel() dispatch. Waiting here would deadlock the
-            // dispatcher, so suppress that not-yet-invoked callback.
-            dispatch_phase = phase::unregistered;
-            invoking_thread = {};
-            dispatcher_identity = nullptr;
-            lock.unlock();
-            destroy_payload();
-            dispatch_cv.notify_all();
-            return;
-        }
+            if (observed == phase::claimed &&
+                dispatcher_identity == current_callback_dispatcher) {
+                // A callback may unregister a later callback selected by the
+                // same synchronous cancel() dispatch. Waiting here would
+                // deadlock the dispatcher, so suppress that not-yet-invoked
+                // callback.
+                if (dispatch_phase.compare_exchange_weak(
+                        observed, phase::unregistered,
+                        std::memory_order_acq_rel,
+                        std::memory_order_acquire)) {
+                    destroy_payload();
+                    dispatch_phase.notify_all();
+                    return;
+                }
+                continue;
+            }
 
-        if (dispatch_phase == phase::claimed ||
-            dispatch_phase == phase::invoking) {
             if (current_callback_dispatcher != nullptr) {
                 // Waiting from one cancellation callback for another dispatcher
                 // can form a cross-source wait cycle. Dispatcher ownership keeps
@@ -200,12 +242,26 @@ struct callback_node {
                 // and let every callback selected by that dispatcher complete.
                 return;
             }
-            if (invoking_thread == std::this_thread::get_id()) return;
-            dispatch_cv.wait(lock, [this] {
-                return dispatch_phase != phase::claimed &&
-                       dispatch_phase != phase::invoking;
-            });
+
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+            if (!announced_waiter) {
+                announced_waiter = true;
+                cancel_callback_waiters_for_test.fetch_add(
+                    1, std::memory_order_release);
+                cancel_callback_waiters_for_test.notify_all();
+            }
+#endif
+            dispatch_phase.wait(observed, std::memory_order_acquire);
+            observed = dispatch_phase.load(std::memory_order_acquire);
         }
+
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        if (announced_waiter) {
+            cancel_callback_waiters_for_test.fetch_sub(
+                1, std::memory_order_release);
+            cancel_callback_waiters_for_test.notify_all();
+        }
+#endif
     }
 
     ~callback_node() {
@@ -213,10 +269,7 @@ struct callback_node {
     }
 
 private:
-    std::mutex dispatch_mutex;
-    std::condition_variable dispatch_cv;
-    phase dispatch_phase = phase::registered;
-    std::thread::id invoking_thread;
+    std::atomic<phase> dispatch_phase{phase::registered};
     const void* dispatcher_identity = nullptr;
 };
 
@@ -231,9 +284,10 @@ struct task_parent_callback_node final : callback_node {
 /// Shared cancellation state (implementation detail).
 ///
 /// Stores active callbacks as an intrusive shared-ownership list guarded by a
-/// mutex. Each node also synchronizes callback dispatch with unregistration so
-/// a registration can be destroyed concurrently without releasing callback
-/// captures while they are still in use. Compared with the previous
+/// mutex. Each node uses an atomic dispatch phase to synchronize callback
+/// dispatch with unregistration so a registration can be destroyed
+/// concurrently without releasing callback captures while they are still in
+/// use. Compared with the previous
 /// std::vector<std::pair<id, std::function>> implementation:
 ///   * registration is O(1) without vector growth/copies;
 ///   * each registration owns a single heap-allocated node, and its
@@ -254,6 +308,10 @@ struct cancel_state {
         // Allocate the node before taking the lock to keep the critical
         // section short.
         auto node = std::make_shared<callback_node>();
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        cancel_callback_node_allocations_for_test.fetch_add(
+            1, std::memory_order_relaxed);
+#endif
         node->emplace(std::forward<F>(cb));
 
         if (!add_callback_node(node)) return {};
@@ -320,18 +378,23 @@ struct cancel_state {
                 return;  // Already cancelled
             }
             list = std::move(head);
-            const auto dispatch_thread = std::this_thread::get_id();
             for (auto node = list; node; node = node->next) {
-                node->claim_for_dispatch(
-                    dispatch_thread, &dispatcher_identity);
+                node->claim_for_dispatch(&dispatcher_identity);
             }
         }
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        if (list) {
+            pause_cancel_callback_dispatch_for_test(
+                pause_cancel_callback_after_claim_for_test,
+                cancel_callback_paused_after_claim_for_test);
+        }
+#endif
         // List is now owned by this call. Invoke each callback outside the
         // state lock. Concurrent remove_callback() calls synchronize through
-        // the individual node. Teardown suppresses a not-yet-started callback,
-        // waits for an in-progress callback outside callback dispatch, or
-        // defers cross-dispatch callback reentry without suppressing callbacks
-        // already selected by another dispatcher.
+        // the individual node's atomic phase. Teardown suppresses a
+        // not-yet-started callback, waits for an in-progress callback outside
+        // callback dispatch, or defers cross-dispatch callback reentry without
+        // suppressing callbacks already selected by another dispatcher.
         std::exception_ptr first_exception;
         while (list) {
             auto node = std::move(list);
@@ -590,6 +653,10 @@ class cancel_token {
         if (!state_) return {};
 
         auto node = std::make_shared<detail::task_parent_callback_node>();
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        detail::cancel_callback_node_allocations_for_test.fetch_add(
+            1, std::memory_order_relaxed);
+#endif
         node->parent_state = state_;
         std::weak_ptr<detail::task_parent_callback_node> weak_node = node;
         node->emplace(
diff --git a/tests/unit/test_cancel_token.cpp b/tests/unit/test_cancel_token.cpp
index ab03fb08..0a5c4b31 100644
--- a/tests/unit/test_cancel_token.cpp
+++ b/tests/unit/test_cancel_token.cpp
@@ -5,7 +5,9 @@
 #include <elio/time/timer.hpp>
 
 #include <atomic>
+#include <array>
 #include <barrier>
+#include <cstddef>
 #include <latch>
 #include <memory>
 #include <optional>
@@ -55,6 +57,73 @@ struct unregister_on_destroy {
     }
 };
 
+template<std::size_t PayloadBytes>
+struct counted_callback {
+    std::atomic<unsigned>* invocations;
+    std::atomic<unsigned>* destructions;
+    bool owns_payload = true;
+    std::array<std::byte, PayloadBytes> payload{};
+
+    counted_callback(std::atomic<unsigned>* invocation_count,
+                     std::atomic<unsigned>* destruction_count) noexcept
+        : invocations(invocation_count), destructions(destruction_count) {}
+
+    counted_callback(counted_callback&& other) noexcept
+        : invocations(other.invocations),
+          destructions(other.destructions),
+          owns_payload(std::exchange(other.owns_payload, false)),
+          payload(other.payload) {}
+
+    counted_callback(const counted_callback&) = delete;
+    counted_callback& operator=(const counted_callback&) = delete;
+
+    ~counted_callback() {
+        if (owns_payload) {
+            destructions->fetch_add(1, std::memory_order_relaxed);
+        }
+    }
+
+    void operator()() {
+        invocations->fetch_add(1, std::memory_order_relaxed);
+    }
+};
+
+struct release_on_destroy_callback {
+    std::atomic<bool>* invoked;
+    std::atomic<bool>* destroyed;
+    bool owns_payload = true;
+
+    release_on_destroy_callback(std::atomic<bool>* invocation_flag,
+                                std::atomic<bool>* destruction_flag) noexcept
+        : invoked(invocation_flag), destroyed(destruction_flag) {}
+    release_on_destroy_callback(release_on_destroy_callback&& other) noexcept
+        : invoked(other.invoked),
+          destroyed(other.destroyed),
+          owns_payload(std::exchange(other.owns_payload, false)) {}
+    release_on_destroy_callback(const release_on_destroy_callback&) = delete;
+    ~release_on_destroy_callback() {
+        if (owns_payload) destroyed->store(true, std::memory_order_release);
+    }
+    void operator()() { invoked->store(true, std::memory_order_release); }
+};
+
+void wait_for_test_hook(std::atomic<bool>& hook) {
+    while (!hook.load(std::memory_order_acquire)) {
+        hook.wait(false, std::memory_order_acquire);
+    }
+}
+
+void wait_for_cancel_callback_waiter() {
+    auto count = elio::coro::detail::cancel_callback_waiters_for_test.load(
+        std::memory_order_acquire);
+    while (count == 0) {
+        elio::coro::detail::cancel_callback_waiters_for_test.wait(
+            count, std::memory_order_acquire);
+        count = elio::coro::detail::cancel_callback_waiters_for_test.load(
+            std::memory_order_acquire);
+    }
+}
+
 
 
 // ==================== Basic Operations ====================
@@ -159,6 +228,18 @@ TEST_CASE("cancel_source multiple cancel calls are safe", "[cancel_token][source
 
 // ==================== Callbacks ====================
 
+TEST_CASE("cancel callback phase uses lock-free native atomic storage",
+          "[cancel_token][callback][layout]") {
+    static_assert(noexcept(
+        std::declval<cancel_registration&>().unregister()));
+    std::atomic<elio::coro::detail::callback_node::phase> phase;
+    INFO("callback_node bytes: "
+         << sizeof(elio::coro::detail::callback_node));
+    INFO("task_parent_callback_node bytes: "
+         << sizeof(elio::coro::detail::task_parent_callback_node));
+    REQUIRE(phase.is_lock_free());
+}
+
 TEST_CASE("cancel_token on_cancel registers callback", "[cancel_token][callback]") {
     cancel_source source;
     cancel_token token = source.get_token();
@@ -217,6 +298,154 @@ TEST_CASE("cancel_token multiple callbacks all invoked", "[cancel_token][callbac
     }
 }
 
+TEST_CASE("cancel callbacks retain LIFO dispatch order",
+          "[cancel_token][callback][ordering]") {
+    cancel_source source;
+    std::vector<unsigned> order;
+
+    auto first = source.get_token().on_cancel([&] { order.push_back(1); });
+    auto second = source.get_token().on_cancel([&] { order.push_back(2); });
+    auto third = source.get_token().on_cancel([&] { order.push_back(3); });
+
+    source.cancel();
+
+    REQUIRE(order == std::vector<unsigned>{3, 2, 1});
+}
+
+TEST_CASE("cancel callback payload is destroyed exactly once",
+          "[cancel_token][callback][lifetime]") {
+    std::atomic<unsigned> invocations{0};
+    std::atomic<unsigned> destructions{0};
+    elio::coro::detail::cancel_callback_node_allocations_for_test.store(
+        0, std::memory_order_relaxed);
+
+    SECTION("inline payload after dispatch") {
+        cancel_source source;
+        auto registration = source.get_token().on_cancel(
+            counted_callback<1>{&invocations, &destructions});
+
+        REQUIRE(elio::coro::detail::cancel_callback_node_allocations_for_test.load(
+                    std::memory_order_relaxed) == 1);
+        source.cancel();
+
+        REQUIRE(invocations.load(std::memory_order_relaxed) == 1);
+        REQUIRE(destructions.load(std::memory_order_relaxed) == 1);
+    }
+
+    SECTION("heap payload after unregister") {
+        cancel_source source;
+        auto registration = source.get_token().on_cancel(
+            counted_callback<96>{&invocations, &destructions});
+
+        REQUIRE(elio::coro::detail::cancel_callback_node_allocations_for_test.load(
+                    std::memory_order_relaxed) == 1);
+        registration.unregister();
+        source.cancel();
+
+        REQUIRE(invocations.load(std::memory_order_relaxed) == 0);
+        REQUIRE(destructions.load(std::memory_order_relaxed) == 1);
+    }
+
+    SECTION("heap payload after immediate cancellation") {
+        cancel_source source;
+        source.cancel();
+
+        auto registration = source.get_token().on_cancel(
+            counted_callback<96>{&invocations, &destructions});
+
+        REQUIRE(elio::coro::detail::cancel_callback_node_allocations_for_test.load(
+                    std::memory_order_relaxed) == 1);
+        REQUIRE(invocations.load(std::memory_order_relaxed) == 1);
+        REQUIRE(destructions.load(std::memory_order_relaxed) == 1);
+    }
+}
+
+TEST_CASE("throwing cancel callback destroys its counted payload once",
+          "[cancel_token][callback][exception][lifetime]") {
+    struct throwing_counted_callback {
+        std::atomic<unsigned>* invocations;
+        std::atomic<unsigned>* destructions;
+        bool owns_payload = true;
+
+        throwing_counted_callback(
+            std::atomic<unsigned>* invocation_count,
+            std::atomic<unsigned>* destruction_count) noexcept
+            : invocations(invocation_count),
+              destructions(destruction_count) {}
+        throwing_counted_callback(throwing_counted_callback&& other) noexcept
+            : invocations(other.invocations),
+              destructions(other.destructions),
+              owns_payload(std::exchange(other.owns_payload, false)) {}
+        throwing_counted_callback(const throwing_counted_callback&) = delete;
+        ~throwing_counted_callback() {
+            if (owns_payload) {
+                destructions->fetch_add(1, std::memory_order_relaxed);
+            }
+        }
+        void operator()() {
+            invocations->fetch_add(1, std::memory_order_relaxed);
+            throw std::runtime_error("counted callback failed");
+        }
+    };
+
+    std::atomic<unsigned> invocations{0};
+    std::atomic<unsigned> destructions{0};
+
+    SECTION("selected callback") {
+        cancel_source source;
+        auto registration = source.get_token().on_cancel(
+            throwing_counted_callback{&invocations, &destructions});
+
+        std::string exception_message;
+        try {
+            source.cancel();
+        } catch (const std::runtime_error& exception) {
+            exception_message = exception.what();
+        }
+        REQUIRE(exception_message == "counted callback failed");
+        REQUIRE(invocations.load(std::memory_order_relaxed) == 1);
+        REQUIRE(destructions.load(std::memory_order_relaxed) == 1);
+    }
+
+    SECTION("already-cancelled immediate callback") {
+        cancel_source source;
+        source.cancel();
+
+        std::string exception_message;
+        try {
+            auto registration = source.get_token().on_cancel(
+                throwing_counted_callback{&invocations, &destructions});
+            (void)registration;
+        } catch (const std::runtime_error& exception) {
+            exception_message = exception.what();
+        }
+        REQUIRE(exception_message == "counted callback failed");
+        REQUIRE(invocations.load(std::memory_order_relaxed) == 1);
+        REQUIRE(destructions.load(std::memory_order_relaxed) == 1);
+    }
+}
+
+TEST_CASE("concurrent callback-node teardown destroys its payload once",
+          "[cancel_token][callback][thread][lifetime]") {
+    auto state = std::make_shared<elio::coro::detail::cancel_state>();
+    auto node = std::make_shared<elio::coro::detail::callback_node>();
+    std::atomic<unsigned> invocations{0};
+    std::atomic<unsigned> destructions{0};
+    node->emplace(counted_callback<1>{&invocations, &destructions});
+    REQUIRE(state->add_callback_node(node));
+
+    std::vector<std::thread> teardown_threads;
+    for (unsigned i = 0; i < 4; ++i) {
+        teardown_threads.emplace_back(
+            [state, node] { state->remove_callback(node); });
+    }
+    for (auto& thread : teardown_threads) thread.join();
+
+    state->trigger();
+    REQUIRE(invocations.load(std::memory_order_relaxed) == 0);
+    REQUIRE(destructions.load(std::memory_order_relaxed) == 1);
+}
+
 TEST_CASE("cancel_token releases all callbacks when one throws",
           "[cancel_token][callback][regression]") {
     cancel_source source;
@@ -745,80 +974,103 @@ TEST_CASE("cancel_token stress test with many callbacks", "[cancel_token][thread
 TEST_CASE("cancel registration waits for a callback running on another thread",
           "[cancel_token][callback][thread][lifetime]") {
     cancel_source source;
-    std::latch callback_started(1);
-    std::latch release_callback(1);
-    std::latch unregister_started(1);
     std::atomic<bool> unregister_done{false};
+    std::atomic<bool> payload_destroyed_at_unregister_return{false};
     std::atomic<bool> callback_done{false};
+    std::atomic<bool> payload_destroyed{false};
+    elio::coro::detail::cancel_callback_waiters_for_test.store(
+        0, std::memory_order_relaxed);
+    elio::coro::detail::cancel_callback_paused_after_invoking_for_test.store(
+        false, std::memory_order_relaxed);
+    elio::coro::detail::pause_cancel_callback_after_invoking_for_test.store(
+        true, std::memory_order_release);
 
-    auto registration = source.get_token().on_cancel([&] {
-        callback_started.count_down();
-        release_callback.wait();
-        callback_done.store(true, std::memory_order_release);
-    });
+    auto registration = source.get_token().on_cancel(
+        release_on_destroy_callback{&callback_done, &payload_destroyed});
 
     std::thread canceller([&] { source.cancel(); });
-    callback_started.wait();
+    wait_for_test_hook(
+        elio::coro::detail::cancel_callback_paused_after_invoking_for_test);
 
     std::thread unregisterer(
-        [registration = std::move(registration), &unregister_started,
+        [registration = std::move(registration), &payload_destroyed,
+         &payload_destroyed_at_unregister_return,
          &unregister_done]() mutable {
-            unregister_started.count_down();
             registration.unregister();
+            payload_destroyed_at_unregister_return.store(
+                payload_destroyed.load(std::memory_order_acquire),
+                std::memory_order_release);
             unregister_done.store(true, std::memory_order_release);
         });
-    unregister_started.wait();
+    wait_for_cancel_callback_waiter();
 
-    // Give the dedicated unregister thread a scheduling opportunity. It must
-    // remain inside unregister() until the callback has stopped using captures.
-    std::this_thread::sleep_for(std::chrono::milliseconds(20));
-    REQUIRE_FALSE(unregister_done.load(std::memory_order_acquire));
+    CHECK_FALSE(unregister_done.load(std::memory_order_acquire));
+    CHECK_FALSE(callback_done.load(std::memory_order_acquire));
 
-    release_callback.count_down();
+    elio::coro::detail::pause_cancel_callback_after_invoking_for_test.store(
+        false, std::memory_order_release);
+    elio::coro::detail::pause_cancel_callback_after_invoking_for_test
+        .notify_all();
     canceller.join();
     unregisterer.join();
     REQUIRE(callback_done.load(std::memory_order_acquire));
+    REQUIRE(payload_destroyed.load(std::memory_order_acquire));
+    REQUIRE(payload_destroyed_at_unregister_return.load(
+        std::memory_order_acquire));
     REQUIRE(unregister_done.load(std::memory_order_acquire));
+    REQUIRE(elio::coro::detail::cancel_callback_waiters_for_test.load(
+                std::memory_order_acquire) == 0);
 }
 
 TEST_CASE("cancel registration waits after another thread selects its callback",
           "[cancel_token][callback][thread][lifetime]") {
     cancel_source source;
-    std::latch dispatch_blocked(1);
-    std::latch release_dispatch(1);
-    std::latch unregister_started(1);
     std::atomic<bool> selected_callback_ran{false};
+    std::atomic<bool> payload_destroyed{false};
     std::atomic<bool> unregister_done{false};
-
-    // The list is LIFO. Register the selected callback first so the blocker is
-    // dispatched before it after cancel() atomically claims both callbacks.
-    auto selected = source.get_token().on_cancel([&] {
-        selected_callback_ran.store(true, std::memory_order_release);
-    });
-    auto blocker = source.get_token().on_cancel([&] {
-        dispatch_blocked.count_down();
-        release_dispatch.wait();
-    });
+    std::atomic<bool> payload_destroyed_at_unregister_return{false};
+    elio::coro::detail::cancel_callback_waiters_for_test.store(
+        0, std::memory_order_relaxed);
+    elio::coro::detail::cancel_callback_paused_after_claim_for_test.store(
+        false, std::memory_order_relaxed);
+    elio::coro::detail::pause_cancel_callback_after_claim_for_test.store(
+        true, std::memory_order_release);
+
+    auto selected = source.get_token().on_cancel(
+        release_on_destroy_callback{&selected_callback_ran,
+                                    &payload_destroyed});
 
     std::thread canceller([&] { source.cancel(); });
-    dispatch_blocked.wait();
+    wait_for_test_hook(
+        elio::coro::detail::cancel_callback_paused_after_claim_for_test);
 
     std::thread unregisterer(
-        [selected = std::move(selected), &unregister_started,
+        [selected = std::move(selected), &payload_destroyed,
+         &payload_destroyed_at_unregister_return,
          &unregister_done]() mutable {
-            unregister_started.count_down();
             selected.unregister();
+            payload_destroyed_at_unregister_return.store(
+                payload_destroyed.load(std::memory_order_acquire),
+                std::memory_order_release);
             unregister_done.store(true, std::memory_order_release);
         });
-    unregister_started.wait();
-    std::this_thread::sleep_for(std::chrono::milliseconds(20));
-    REQUIRE_FALSE(unregister_done.load(std::memory_order_acquire));
+    wait_for_cancel_callback_waiter();
+
+    CHECK_FALSE(unregister_done.load(std::memory_order_acquire));
+    CHECK_FALSE(selected_callback_ran.load(std::memory_order_acquire));
 
-    release_dispatch.count_down();
+    elio::coro::detail::pause_cancel_callback_after_claim_for_test.store(
+        false, std::memory_order_release);
+    elio::coro::detail::pause_cancel_callback_after_claim_for_test.notify_all();
     canceller.join();
     unregisterer.join();
     REQUIRE(selected_callback_ran.load(std::memory_order_acquire));
+    REQUIRE(payload_destroyed.load(std::memory_order_acquire));
+    REQUIRE(payload_destroyed_at_unregister_return.load(
+        std::memory_order_acquire));
     REQUIRE(unregister_done.load(std::memory_order_acquire));
+    REQUIRE(elio::coro::detail::cancel_callback_waiters_for_test.load(
+                std::memory_order_acquire) == 0);
 }
 
 TEST_CASE("cancel callback can unregister itself without deadlock",
@@ -837,6 +1089,26 @@ TEST_CASE("cancel callback can unregister itself without deadlock",
     REQUIRE_FALSE(registration.has_value());
 }
 
+TEST_CASE("cancel callback can reenter its own source",
+          "[cancel_token][callback][reentrant]") {
+    cancel_source source;
+    std::atomic<unsigned> reentrant_invocations{0};
+    std::atomic<unsigned> later_invocations{0};
+
+    auto later = source.get_token().on_cancel([&] {
+        later_invocations.fetch_add(1, std::memory_order_relaxed);
+    });
+    auto reentrant = source.get_token().on_cancel([&] {
+        reentrant_invocations.fetch_add(1, std::memory_order_relaxed);
+        source.cancel();
+    });
+
+    source.cancel();
+
+    REQUIRE(reentrant_invocations.load(std::memory_order_relaxed) == 1);
+    REQUIRE(later_invocations.load(std::memory_order_relaxed) == 1);
+}
+
 TEST_CASE("cancel callback can remove a later selected callback",
           "[cancel_token][callback][thread][lifetime]") {
     cancel_source source;
@@ -857,6 +1129,25 @@ TEST_CASE("cancel callback can remove a later selected callback",
     REQUIRE_FALSE(later_ran.load(std::memory_order_acquire));
 }
 
+TEST_CASE("same dispatcher suppresses a claimed counted callback exactly once",
+          "[cancel_token][callback][lifetime]") {
+    cancel_source source;
+    std::optional<cancel_registration> later_registration;
+    std::atomic<unsigned> invocations{0};
+    std::atomic<unsigned> destructions{0};
+
+    later_registration.emplace(source.get_token().on_cancel(
+        counted_callback<1>{&invocations, &destructions}));
+    auto first_registration = source.get_token().on_cancel(
+        [&] { later_registration.reset(); });
+
+    source.cancel();
+
+    REQUIRE(invocations.load(std::memory_order_relaxed) == 0);
+    REQUIRE(destructions.load(std::memory_order_relaxed) == 1);
+    REQUIRE_FALSE(later_registration.has_value());
+}
+
 TEST_CASE("nested cancellation preserves an outer selected callback",
           "[cancel_token][callback][lifetime]") {
     cancel_source outer_source;
@@ -887,6 +1178,34 @@ TEST_CASE("nested cancellation preserves an outer selected callback",
     REQUIRE_FALSE(outer_later_registration.has_value());
 }
 
+TEST_CASE("immediate nested cancellation preserves an outer selected callback",
+          "[cancel_token][callback][lifetime]") {
+    cancel_source outer_source;
+    cancel_source already_cancelled_source;
+    already_cancelled_source.cancel();
+    std::optional<cancel_registration> outer_later_registration;
+    std::atomic<unsigned> outer_later_invocations{0};
+    std::atomic<bool> immediate_ran{false};
+
+    outer_later_registration.emplace(
+        outer_source.get_token().on_cancel([&] {
+            outer_later_invocations.fetch_add(1, std::memory_order_relaxed);
+        }));
+    auto outer_first_registration = outer_source.get_token().on_cancel([&] {
+        auto immediate_registration =
+            already_cancelled_source.get_token().on_cancel([&] {
+                immediate_ran.store(true, std::memory_order_release);
+                outer_later_registration.reset();
+            });
+    });
+
+    outer_source.cancel();
+
+    REQUIRE(immediate_ran.load(std::memory_order_acquire));
+    REQUIRE(outer_later_invocations.load(std::memory_order_relaxed) == 1);
+    REQUIRE_FALSE(outer_later_registration.has_value());
+}
+
 TEST_CASE("callbacks on separate dispatchers can mutually unregister",
           "[cancel_token][callback][thread][lifetime]") {
     cancel_source first_source;
diff --git a/wiki/Performance-Tuning.md b/wiki/Performance-Tuning.md
index 44909ad5..9422c92c 100644
--- a/wiki/Performance-Tuning.md
+++ b/wiki/Performance-Tuning.md
@@ -403,6 +403,43 @@ Do not eagerly invoke an arbitrary temporary coroutine lambda merely to select
 the direct overload: its returned frame may retain the lambda through `this`.
 Passing the callable lets Elio keep that object alive in the wrapper.
 
+### Cancellation Callback Registration
+
+Each `cancel_token::on_cancel()` registration owns one shared callback node.
+Small nothrow-movable callables use the node's inline buffer; larger callables
+require a second payload allocation. The node uses a native-width atomic phase
+for selection, invocation, and teardown. Ordinary register/unregister traffic
+therefore does not initialize or lock a per-registration mutex or condition
+variable. The cancellation-state mutex still owns list insertion, selection,
+and O(N) unlinking, so unregistering an old callback from a long-lived source
+costs more than removing its newest callback.
+
+`cancel_callback_benchmark` reports the platform-specific node sizes and
+allocator-requested bytes for an inline callback, register/unregister latency,
+newest and oldest unlink latency for several list sizes, cancellation dispatch,
+immediate registration after cancellation, and cancellation p95/p99 while
+another thread unregisters callbacks. Build it in Release mode and collect
+multiple process-level runs on a fixed CPU when comparing revisions:
+
+```bash
+cmake -S . -B build-release -DCMAKE_BUILD_TYPE=Release \
+  -DELIO_BUILD_EXAMPLES=ON
+cmake --build build-release --target cancel_callback_benchmark --parallel 2
+taskset -c 2 ./build-release/examples/cancel_callback_benchmark 2000
+```
+
+Use `--smoke` for a short 20-sample termination and output check. The optional
+numeric argument must be a strict positive sample count; invalid or additional
+arguments return exit status 2.
+
+Use at least two isolated CPUs for the concurrent-unregister tail diagnostic,
+for example `taskset -c 2,3`. Its `removed` and `invoked` outcome totals confirm
+that cancellation and teardown actually overlapped. Keep the one-CPU run for
+the single-thread register, unlink, dispatch, and immediate-callback metrics.
+Use an external paired runner across multiple process invocations to calculate
+revision-to-revision confidence intervals; the in-process percentiles are
+diagnostic samples, not a substitute for paired confidence intervals.
+
 ### Avoiding Allocations
 
 Keep coroutine frames small to reduce allocation and cache cost: