Merge branch 'pratik/std-coro/migrate-test-code' into pratik/std-coro/cleanup-boost-coroutine

Bring in develop via the upstream branch. Two classes of collision needed
manual resolution:

Namespace case refactor (develop #7933) renamed RPC -> rpc and
xrpl::RPC::Tuning -> xrpl::rpc::tuning. Conflicts were resolved by taking
develop's naming together with this stack's coroutine API (postCoroTask,
CoroTask<void>, no Context::coro):
  - src/test/app/Path_test.cpp
  - src/test/app/PathMPT_test.cpp
  - src/test/jtx/impl/TestHelpers.cpp
  - src/xrpld/rpc/handlers/orderbook/RipplePathFind.cpp (no textual
    conflict; stale spellings carried forward on stack-authored lines)

New old-API caller: develop added postCoro and Context::coro uses in
src/test/app/PayChan_test.cpp, which this stack does not otherwise touch,
so it merged without conflict but would not compile once Context::coro is
removed here. Both call sites are migrated to postCoroTask and the
.coro = {} designated initialisers dropped. The kFeeHeavyBurdenRpc
assertions that motivated develop's change are preserved unchanged.

.cspell.config.yaml

@@ -256,6 +256,8 @@ words:
   - replayer
   - repodata
   - repomd
+  - repost
+  - reposts
   - rerandomize
   - rerandomization
   - rerandomized
@@ -303,6 +305,7 @@ words:
   - sponsees
   - SRPMS
   - sslws
+  - stackful
   - statsd
   - STATSDCOLLECTOR
   - stissue

cmake/deps/Boost.cmake

@@ -39,21 +39,12 @@ if(Boost_COMPILER)
     target_link_libraries(xrpl_boost INTERFACE Boost::disable_autolinking)
 endif()
 
-# GCC 14+ has a false positive -Wuninitialized warning in Boost.Coroutine2's
-# state.hpp when compiled with -O3. This is due to GCC's intentional behavior
-# change (Bug #98871, #119388) where warnings from inlined system header code
-# are no longer suppressed by -isystem. The warning occurs in operator|= in
-# boost/coroutine2/detail/state.hpp when inlined from push_control_block::destroy().
-# See: https://gcc.gnu.org/bugzilla/show_bug.cgi?id=119388
-if(is_gcc AND CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL 14)
-    target_compile_options(xrpl_boost INTERFACE -Wno-uninitialized)
-endif()
-
 # Boost.Context's ucontext backend has ASAN fiber-switching annotations
 # (start/finish_switch_fiber) that are compiled in when BOOST_USE_ASAN is defined.
-# This tells ASAN about coroutine stack switches, preventing false positive
-# stack-use-after-scope errors. BOOST_USE_UCONTEXT ensures the ucontext backend
-# is selected (fcontext does not support ASAN annotations).
+# This tells ASAN about fiber stack switches used by boost::asio::spawn,
+# preventing false positive stack-use-after-scope errors.
+# BOOST_USE_UCONTEXT ensures the ucontext backend is selected (fcontext does
+# not support ASAN annotations).
 # These defines must match what Boost was compiled with (see conan/profiles/sanitizers).
 if(enable_asan)
     target_compile_definitions(

conanfile.py

@@ -62,7 +62,7 @@ class Xrpl(ConanFile):
         "boost/*:without_cobalt": True,
         "boost/*:without_context": False,
         "boost/*:without_coroutine": True,
-        "boost/*:without_coroutine2": False,
+        "boost/*:without_coroutine2": True,
         "date/*:header_only": True,
         "ed25519/*:shared": False,
         "grpc/*:shared": False,

include/xrpl/core/Coro.ipp

@@ -1,136 +0,0 @@
-#pragma once
-
-#include <utility>
-
-namespace xrpl {
-
-/**
- * Coroutine stack size (1.5 MB). Increased from 1 MB because
- * ASAN-instrumented deep call stacks exceeded the original limit.
- */
-constexpr std::size_t kCoroStackSize = 1536 * 1024;
-
-template <class F>
-JobQueue::Coro::Coro(CoroCreateT, JobQueue& jq, JobType type, std::string name, F&& f)
-    : jq_(jq)
-    , type_(type)
-    , name_(std::move(name))
-    , coro_(
-          boost::context::protected_fixedsize_stack(kCoroStackSize),
-          [this, fn = std::forward<F>(f)](boost::coroutines2::coroutine<void>::push_type& doYield) {
-              yield_ = &doYield;
-              yield();
-              fn(shared_from_this());
-#ifndef NDEBUG
-              finished_ = true;
-#endif
-          })
-{
-}
-
-inline JobQueue::Coro::~Coro()
-{
-#ifndef NDEBUG
-    XRPL_ASSERT(finished_, "xrpl::JobQueue::Coro::~Coro : is finished");
-#endif
-}
-
-inline void
-JobQueue::Coro::yield() const
-{
-    {
-        std::scoped_lock const lock(jq_.mutex_);
-        ++jq_.nSuspend_;
-    }
-    (*yield_)();
-}
-
-inline bool
-JobQueue::Coro::post()
-{
-    {
-        std::scoped_lock const lk(mutexRun_);
-        running_ = true;
-    }
-
-    // sp keeps 'this' alive
-    if (jq_.addJob(type_, name_, [this, sp = shared_from_this()]() { resume(); }))
-    {
-        return true;
-    }
-
-    // The coroutine will not run.  Clean up running_.
-    std::scoped_lock const lk(mutexRun_);
-    running_ = false;
-    cv_.notify_all();
-    return false;
-}
-
-inline void
-JobQueue::Coro::resume()
-{
-    {
-        std::scoped_lock const lk(mutexRun_);
-        running_ = true;
-    }
-    {
-        std::scoped_lock const lk(jq_.mutex_);
-        --jq_.nSuspend_;
-    }
-    auto saved = detail::getLocalValues().release();
-    detail::getLocalValues().reset(&lvs_);
-    std::scoped_lock const lock(mutex_);
-    // A late resume() can arrive after the coroutine has already completed.
-    // This is an expected (if rare) outcome of the race condition documented
-    // in JobQueue.h:354-377 where post() schedules a resume job before the
-    // coroutine yields — the mutex serializes access, but by the time this
-    // resume() acquires the lock the coroutine may have already run to
-    // completion. Calling operator() on a completed boost::coroutine2 is
-    // undefined behavior, so we must check and skip invoking the coroutine
-    // body if it has already completed.
-    if (coro_)
-    {
-        coro_();
-    }
-    detail::getLocalValues().release();
-    detail::getLocalValues().reset(saved);
-    std::scoped_lock const lk(mutexRun_);
-    running_ = false;
-    cv_.notify_all();
-}
-
-inline bool
-JobQueue::Coro::runnable() const
-{
-    return static_cast<bool>(coro_);
-}
-
-inline void
-JobQueue::Coro::expectEarlyExit()
-{
-#ifndef NDEBUG
-    if (!finished_)
-#endif
-    {
-        // expectEarlyExit() must only ever be called from outside the
-        // Coro's stack.  It you're inside the stack you can simply return
-        // and be done.
-        //
-        // That said, since we're outside the Coro's stack, we need to
-        // decrement the nSuspend that the Coro's call to yield caused.
-        std::scoped_lock const lock(jq_.mutex_);
-        --jq_.nSuspend_;
-#ifndef NDEBUG
-        finished_ = true;
-#endif
-    }
-}
-
-inline void
-JobQueue::Coro::join()
-{
-    std::unique_lock<std::mutex> lk(mutexRun_);
-    cv_.wait(lk, [this]() { return !running_; });
-}
-
-}  // namespace xrpl

include/xrpl/core/CoroTask.h

@@ -0,0 +1,713 @@
+#pragma once
+
+#include <xrpl/beast/utility/instrumentation.h>
+
+#include <coroutine>
+#include <exception>
+#include <type_traits>
+#include <utility>
+#include <variant>
+
+namespace xrpl {
+
+template <typename T = void>
+class CoroTask;
+
+/**
+ * CoroTask<void> -- coroutine return type for void-returning coroutines.
+ *
+ * Class / Dependency Diagram
+ * ==========================
+ *
+ *   CoroTask<void>
+ *   +-----------------------------------------------+
+ *   | - handle_ : Handle (coroutine_handle<promise>) |
+ *   +-----------------------------------------------+
+ *   | + handle(), done()                             |
+ *   | + await_ready/suspend/resume  (Awaiter iface)  |
+ *   +-----------------------------------------------+
+ *            |  owns
+ *            v
+ *   promise_type
+ *   +-----------------------------------------------+
+ *   | - exception_    : std::exception_ptr           |
+ *   | - continuation_ : std::coroutine_handle<>      |
+ *   +-----------------------------------------------+
+ *   | + get_return_object() -> CoroTask              |
+ *   | + initial_suspend()   -> suspend_always (lazy) |
+ *   | + final_suspend()     -> FinalAwaiter          |
+ *   | + return_void()                                |
+ *   | + unhandled_exception()                        |
+ *   +-----------------------------------------------+
+ *            |  returns at final_suspend
+ *            v
+ *   FinalAwaiter
+ *   +-----------------------------------------------+
+ *   | await_suspend(h):                              |
+ *   |   if continuation_ set -> symmetric transfer   |
+ *   |   else                 -> noop_coroutine        |
+ *   +-----------------------------------------------+
+ *
+ * Design Notes
+ * ------------
+ * - Lazy start: initial_suspend returns suspend_always, so the coroutine
+ *   body does not execute until the handle is explicitly resumed.
+ * - Symmetric transfer: await_suspend returns a coroutine_handle instead
+ *   of void/bool, allowing the scheduler to jump directly to the next
+ *   coroutine without growing the call stack.
+ * - Continuation chaining: when one CoroTask is co_await-ed inside
+ *   another, the caller's handle is stored as continuation_ so
+ *   FinalAwaiter can resume it when this task finishes.
+ * - Move-only: the handle is exclusively owned; copy is deleted.
+ *
+ * Usage Examples
+ * ==============
+ *
+ * 1. Basic void coroutine (the most common case in xrpld):
+ *
+ *      CoroTask<void> doWork(std::shared_ptr<CoroTaskRunner> runner) {
+ *          // do something
+ *          co_await runner->suspend();   // yield control
+ *          // resumed later via runner->post() or runner->resume()
+ *          co_return;
+ *      }
+ *
+ * 2. co_await-ing one CoroTask<void> from another (chaining):
+ *
+ *      CoroTask<void> inner() {
+ *          // ...
+ *          co_return;
+ *      }
+ *      CoroTask<void> outer() {
+ *          co_await inner();   // continuation_ links outer -> inner
+ *          co_return;          // FinalAwaiter resumes outer
+ *      }
+ *
+ * 3. Exceptions propagate through co_await:
+ *
+ *      CoroTask<void> failing() {
+ *          throw std::runtime_error("oops");
+ *          co_return;
+ *      }
+ *      CoroTask<void> caller() {
+ *          try { co_await failing(); }
+ *          catch (std::runtime_error const&) { // caught here }
+ *      }
+ *
+ * Caveats / Pitfalls
+ * ==================
+ *
+ * BUG-RISK: Dangling references in coroutine parameters.
+ *   Coroutine parameters are copied into the frame, but references
+ *   are NOT -- they are stored as-is. If the referent goes out of scope
+ *   before the coroutine finishes, you get use-after-free.
+ *
+ *      // BROKEN -- local dies before coroutine runs:
+ *      CoroTask<void> bad(int& ref) { co_return; }
+ *      void launch() {
+ *          int local = 42;
+ *          auto task = bad(local);  // frame stores &local
+ *      }  // local destroyed; frame holds dangling ref
+ *
+ *      // FIX -- pass by value, or ensure lifetime via shared_ptr.
+ *
+ * BUG-RISK: GCC 14 corrupts reference captures in coroutine lambdas.
+ *   When a lambda that returns CoroTask captures by reference ([&]),
+ *   GCC 14 may generate a corrupted coroutine frame. Always capture
+ *   by explicit pointer-to-value instead:
+ *
+ *      // BROKEN on GCC 14:
+ *      jq.postCoroTask(t, n, [&](auto) -> CoroTask<void> { ... });
+ *
+ *      // FIX -- capture pointers explicitly:
+ *      jq.postCoroTask(t, n, [ptr = &val](auto) -> CoroTask<void> { ... });
+ *
+ * BUG-RISK: Resuming a destroyed or completed CoroTask.
+ *   Calling handle().resume() after the coroutine has already run to
+ *   completion (done() == true) is undefined behavior. The CoroTaskRunner
+ *   guards against this with an XRPL_ASSERT, but standalone usage of
+ *   CoroTask must check done() before resuming.
+ *
+ * BUG-RISK: Moving a CoroTask that is being awaited.
+ *   If task A is co_await-ed by task B (so A.continuation_ == B), moving
+ *   or destroying A will invalidate the continuation link. Never move
+ *   or reassign a CoroTask while it is mid-execution or being awaited.
+ *
+ * LIMITATION: CoroTask is fire-and-forget for the top-level owner.
+ *   There is no built-in notification when the coroutine finishes.
+ *   The caller must use external synchronization (e.g. CoroTaskRunner::join
+ *   or a gate/condition_variable) to know when it is done.
+ *
+ * LIMITATION: No cancellation token.
+ *   There is no way to cancel a suspended CoroTask from outside. The
+ *   coroutine body must cooperatively check a flag (e.g. jq_.isStopping())
+ *   after each co_await and co_return early if needed.
+ *
+ * LIMITATION: Stackless -- cannot suspend from nested non-coroutine calls.
+ *   If a coroutine calls a regular function that wants to "yield", it
+ *   cannot. Only the immediate coroutine body can use co_await.
+ *   This is acceptable for xrpld because all yield() sites are shallow.
+ */
+template <>
+class CoroTask<void>
+{
+public:
+    // The C++ coroutine protocol mandates these names (promise_type,
+    // initial_suspend, await_ready, ...) and instance-callable awaiter
+    // methods, which conflict with the project naming/static conventions.
+    // NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
+    struct promise_type;
+    using Handle = std::coroutine_handle<promise_type>;
+
+    /**
+     * Coroutine promise. Compiler uses this to manage coroutine state.
+     * Stores the exception (if any) and the continuation handle for
+     * symmetric transfer back to the awaiting coroutine.
+     */
+    struct promise_type
+    {
+        // Captured exception from the coroutine body, rethrown in
+        // await_resume() when this task is co_await-ed by a caller.
+        std::exception_ptr exception_;
+
+        // Handle to the coroutine that is co_await-ing this task.
+        // Set by await_suspend(). FinalAwaiter uses it for symmetric
+        // transfer back to the caller. Null if this is a top-level task.
+        std::coroutine_handle<> continuation_;
+
+        /**
+         * Create the CoroTask return object.
+         * Called by the compiler at coroutine creation.
+         */
+        CoroTask
+        get_return_object()
+        {
+            return CoroTask{Handle::from_promise(*this)};
+        }
+
+        /**
+         * Lazy start. The coroutine body does not execute until the
+         * handle is explicitly resumed (e.g. by CoroTaskRunner::resume).
+         */
+        std::suspend_always
+        initial_suspend() noexcept
+        {
+            return {};
+        }
+
+        /**
+         * Awaiter returned by final_suspend(). Uses symmetric transfer:
+         * if a continuation exists, transfers control directly to it
+         * (tail-call, no stack growth). Otherwise returns noop_coroutine
+         * so the coroutine frame stays alive for the owner to destroy.
+         */
+        struct FinalAwaiter
+        {
+            /**
+             * Always false. We need await_suspend to run for
+             * symmetric transfer.
+             */
+            bool
+            await_ready() noexcept
+            {
+                return false;
+            }
+
+            /**
+             * Symmetric transfer: returns the continuation handle so
+             * the compiler emits a tail-call instead of a nested resume.
+             * If no continuation is set, returns noop_coroutine to
+             * suspend at final_suspend without destroying the frame.
+             *
+             * @param h Handle to this completing coroutine
+             *
+             * @return Continuation handle, or noop_coroutine
+             */
+            std::coroutine_handle<>
+            await_suspend(Handle h) noexcept
+            {
+                if (auto cont = h.promise().continuation_)
+                    return cont;
+                return std::noop_coroutine();
+            }
+
+            void
+            await_resume() noexcept
+            {
+            }
+        };
+
+        /**
+         * Returns FinalAwaiter for symmetric transfer at coroutine end.
+         */
+        FinalAwaiter
+        final_suspend() noexcept
+        {
+            return {};
+        }
+
+        /**
+         * Called by the compiler for `co_return;` (void coroutine).
+         */
+        void
+        return_void()
+        {
+        }
+
+        /**
+         * Called by the compiler when an exception escapes the coroutine
+         * body. Captures it for later rethrowing in await_resume().
+         */
+        void
+        unhandled_exception()
+        {
+            exception_ = std::current_exception();
+        }
+    };
+    // NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
+
+    /**
+     * Default constructor. Creates an empty (null handle) task.
+     */
+    CoroTask() = default;
+
+    /**
+     * Takes ownership of a compiler-generated coroutine handle.
+     *
+     * @param h Coroutine handle to own
+     */
+    explicit CoroTask(Handle h) : handle_(h)
+    {
+    }
+
+    /**
+     * Destroys the coroutine frame if this task owns one.
+     */
+    ~CoroTask()
+    {
+        if (handle_)
+            handle_.destroy();
+    }
+
+    /**
+     * Move constructor. Transfers handle ownership, leaves other empty.
+     */
+    CoroTask(CoroTask&& other) noexcept : handle_(std::exchange(other.handle_, {}))
+    {
+    }
+
+    /**
+     * Move assignment. Destroys current frame (if any), takes other's.
+     */
+    CoroTask&
+    operator=(CoroTask&& other) noexcept
+    {
+        if (this != &other)
+        {
+            if (handle_)
+                handle_.destroy();
+            handle_ = std::exchange(other.handle_, {});
+        }
+        return *this;
+    }
+
+    CoroTask(CoroTask const&) = delete;
+    CoroTask&
+    operator=(CoroTask const&) = delete;
+
+    /**
+     * @return The underlying coroutine_handle
+     */
+    [[nodiscard]] Handle
+    handle() const
+    {
+        return handle_;
+    }
+
+    /**
+     * @return true if the coroutine has run to completion (or thrown)
+     */
+    [[nodiscard]] bool
+    done() const
+    {
+        return handle_ && handle_.done();
+    }
+
+    // -- Awaiter interface: allows `co_await someCoroTask;` --
+
+    /**
+     * Always false. This task is lazy, so co_await always suspends
+     * the caller to set up the continuation link.
+     */
+    // NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
+    [[nodiscard]] bool
+    await_ready() const noexcept
+    {
+        return false;
+    }
+
+    /**
+     * Stores the caller's handle as our continuation, then returns
+     * our handle for symmetric transfer (caller suspends, we resume).
+     *
+     * @param caller Handle of the coroutine doing co_await on us
+     *
+     * @return Our handle for symmetric transfer
+     */
+    std::coroutine_handle<>
+    await_suspend(std::coroutine_handle<> caller) noexcept
+    {
+        XRPL_ASSERT(handle_, "xrpl::CoroTask<void>::await_suspend : handle is valid");
+        handle_.promise().continuation_ = caller;
+        return handle_;  // Symmetric transfer
+    }
+
+    /**
+     * Called in the awaiting coroutine's context after this task
+     * completes. Rethrows any exception captured by
+     * unhandled_exception().
+     */
+    void
+    await_resume()
+    {
+        XRPL_ASSERT(handle_, "xrpl::CoroTask<void>::await_resume : handle is valid");
+        if (auto& ep = handle_.promise().exception_)
+            std::rethrow_exception(ep);
+    }
+    // NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
+
+private:
+    // Exclusively-owned coroutine handle. Null after move or default
+    // construction. Destroyed in the destructor.
+    Handle handle_;
+};
+
+/**
+ * CoroTask<T> -- coroutine return type for value-returning coroutines.
+ *
+ * Class / Dependency Diagram
+ * ==========================
+ *
+ *   CoroTask<T>
+ *   +-----------------------------------------------+
+ *   | - handle_ : Handle (coroutine_handle<promise>) |
+ *   +-----------------------------------------------+
+ *   | + handle(), done()                             |
+ *   | + await_ready/suspend/resume  (Awaiter iface)  |
+ *   +-----------------------------------------------+
+ *            |  owns
+ *            v
+ *   promise_type
+ *   +-----------------------------------------------+
+ *   | - result_       : variant<monostate, T,        |
+ *   |                           exception_ptr>       |
+ *   | - continuation_ : std::coroutine_handle<>      |
+ *   +-----------------------------------------------+
+ *   | + get_return_object() -> CoroTask              |
+ *   | + initial_suspend()   -> suspend_always (lazy) |
+ *   | + final_suspend()     -> FinalAwaiter          |
+ *   | + return_value(T)     -> stores in result_[1]  |
+ *   | + unhandled_exception -> stores in result_[2]  |
+ *   +-----------------------------------------------+
+ *            |  returns at final_suspend
+ *            v
+ *   FinalAwaiter  (same symmetric-transfer pattern as CoroTask<void>)
+ *
+ * Value Extraction
+ * ----------------
+ * await_resume() inspects the variant:
+ *   - index 2 (exception_ptr) -> rethrow
+ *   - index 1 (T)             -> return value via move
+ *
+ * Usage Examples
+ * ==============
+ *
+ * 1. Simple value return:
+ *
+ *      CoroTask<int> computeAnswer() { co_return 42; }
+ *
+ *      CoroTask<void> caller() {
+ *          int v = co_await computeAnswer();  // v == 42
+ *      }
+ *
+ * 2. Chaining value-returning coroutines:
+ *
+ *      CoroTask<int> add(int a, int b) { co_return a + b; }
+ *      CoroTask<int> doubleSum(int a, int b) {
+ *          int s = co_await add(a, b);
+ *          co_return s * 2;
+ *      }
+ *
+ * 3. Exception propagation from inner to outer:
+ *
+ *      CoroTask<int> failing() {
+ *          throw std::runtime_error("bad");
+ *          co_return 0;  // never reached
+ *      }
+ *      CoroTask<void> caller() {
+ *          try {
+ *              int v = co_await failing();  // throws here
+ *          } catch (std::runtime_error const& e) {
+ *              // e.what() == "bad"
+ *          }
+ *      }
+ *
+ * Caveats / Pitfalls (in addition to CoroTask<void> caveats above)
+ * ================================================================
+ *
+ * BUG-RISK: await_resume() moves the value out of the variant.
+ *   Calling co_await on the same CoroTask<T> instance twice is undefined
+ *   behavior -- the second call will see a moved-from T. CoroTask is
+ *   single-shot: one co_return, one co_await.
+ *
+ * BUG-RISK: T must be move-constructible.
+ *   return_value(T) takes by value and moves into the variant.
+ *   Types that are not movable cannot be used as T.
+ *
+ * LIMITATION: No co_yield support.
+ *   CoroTask<T> only supports a single co_return. It does not implement
+ *   yield_value(), so using co_yield inside a CoroTask<T> coroutine is a
+ *   compile error. For streaming values, a different return type
+ *   (e.g. Generator<T>) would be needed.
+ *
+ * LIMITATION: Result is only accessible via co_await.
+ *   There is no .get() or .result() method. The value can only be
+ *   extracted by co_await-ing the CoroTask<T> from inside another
+ *   coroutine. For extracting results in non-coroutine code, pass a
+ *   pointer to the caller and write through it (as the tests do).
+ */
+template <typename T>
+class CoroTask
+{
+    static_assert(
+        std::is_move_constructible_v<T>,
+        "CoroTask<T> requires T to be move-constructible");
+
+public:
+    // The C++ coroutine protocol mandates these names (promise_type,
+    // initial_suspend, await_ready, ...) and instance-callable awaiter
+    // methods, which conflict with the project naming/static conventions.
+    // NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
+    struct promise_type;
+    using Handle = std::coroutine_handle<promise_type>;
+
+    /**
+     * Coroutine promise for value-returning coroutines.
+     * Stores the result as a variant: monostate (not yet set),
+     * T (co_return value), or exception_ptr (unhandled exception).
+     */
+    struct promise_type
+    {
+        // Tri-state result:
+        //   index 0 (monostate) -- coroutine has not yet completed
+        //   index 1 (T)         -- co_return value stored here
+        //   index 2 (exception) -- unhandled exception captured here
+        std::variant<std::monostate, T, std::exception_ptr> result_;
+
+        // Handle to the coroutine co_await-ing this task. Used by
+        // FinalAwaiter for symmetric transfer. Null for top-level tasks.
+        std::coroutine_handle<> continuation_;
+
+        /**
+         * Create the CoroTask return object.
+         * Called by the compiler at coroutine creation.
+         */
+        CoroTask
+        get_return_object()
+        {
+            return CoroTask{Handle::from_promise(*this)};
+        }
+
+        /**
+         * Lazy start. Coroutine body does not run until explicitly resumed.
+         */
+        std::suspend_always
+        initial_suspend() noexcept
+        {
+            return {};
+        }
+
+        /**
+         * Symmetric-transfer awaiter at coroutine completion.
+         * Same pattern as CoroTask<void>::FinalAwaiter.
+         */
+        struct FinalAwaiter
+        {
+            bool
+            await_ready() noexcept
+            {
+                return false;
+            }
+
+            /**
+             * Returns continuation for symmetric transfer, or
+             * noop_coroutine if this is a top-level task.
+             *
+             * @param h Handle to this completing coroutine
+             *
+             * @return Continuation handle, or noop_coroutine
+             */
+            std::coroutine_handle<>
+            await_suspend(Handle h) noexcept
+            {
+                if (auto cont = h.promise().continuation_)
+                    return cont;
+                return std::noop_coroutine();
+            }
+
+            void
+            await_resume() noexcept
+            {
+            }
+        };
+
+        FinalAwaiter
+        final_suspend() noexcept
+        {
+            return {};
+        }
+
+        /**
+         * Called by the compiler for `co_return value;`.
+         * Moves the value into result_ at index 1.
+         *
+         * @param value The value to store
+         */
+        void
+        return_value(T value)
+        {
+            result_.template emplace<1>(std::move(value));
+        }
+
+        /**
+         * Captures unhandled exceptions at index 2 of result_.
+         * Rethrown later in await_resume().
+         */
+        void
+        unhandled_exception()
+        {
+            result_.template emplace<2>(std::current_exception());
+        }
+    };
+    // NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
+
+    /**
+     * Default constructor. Creates an empty (null handle) task.
+     */
+    CoroTask() = default;
+
+    /**
+     * Takes ownership of a compiler-generated coroutine handle.
+     *
+     * @param h Coroutine handle to own
+     */
+    explicit CoroTask(Handle h) : handle_(h)
+    {
+    }
+
+    /**
+     * Destroys the coroutine frame if this task owns one.
+     */
+    ~CoroTask()
+    {
+        if (handle_)
+            handle_.destroy();
+    }
+
+    /**
+     * Move constructor. Transfers handle ownership, leaves other empty.
+     */
+    CoroTask(CoroTask&& other) noexcept : handle_(std::exchange(other.handle_, {}))
+    {
+    }
+
+    /**
+     * Move assignment. Destroys current frame (if any), takes other's.
+     */
+    CoroTask&
+    operator=(CoroTask&& other) noexcept
+    {
+        if (this != &other)
+        {
+            if (handle_)
+                handle_.destroy();
+            handle_ = std::exchange(other.handle_, {});
+        }
+        return *this;
+    }
+
+    CoroTask(CoroTask const&) = delete;
+    CoroTask&
+    operator=(CoroTask const&) = delete;
+
+    /**
+     * @return The underlying coroutine_handle
+     */
+    [[nodiscard]] Handle
+    handle() const
+    {
+        return handle_;
+    }
+
+    /**
+     * @return true if the coroutine has run to completion (or thrown)
+     */
+    [[nodiscard]] bool
+    done() const
+    {
+        return handle_ && handle_.done();
+    }
+
+    // -- Awaiter interface: allows `T val = co_await someCoroTask;` --
+
+    /**
+     * Always false. co_await always suspends to set up continuation.
+     */
+    // NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
+    [[nodiscard]] bool
+    await_ready() const noexcept
+    {
+        return false;
+    }
+
+    /**
+     * Stores caller as continuation, returns our handle for
+     * symmetric transfer.
+     *
+     * @param caller Handle of the coroutine doing co_await on us
+     *
+     * @return Our handle for symmetric transfer
+     */
+    std::coroutine_handle<>
+    await_suspend(std::coroutine_handle<> caller) noexcept
+    {
+        XRPL_ASSERT(handle_, "xrpl::CoroTask<T>::await_suspend : handle is valid");
+        handle_.promise().continuation_ = caller;
+        return handle_;
+    }
+
+    /**
+     * Extracts the result: rethrows if exception, otherwise moves
+     * the T value out of the variant. Single-shot: calling twice
+     * on the same task is undefined (moved-from T).
+     *
+     * @return The co_return-ed value
+     */
+    T
+    await_resume()
+    {
+        XRPL_ASSERT(handle_, "xrpl::CoroTask<T>::await_resume : handle is valid");
+        auto& result = handle_.promise().result_;
+        if (auto* ep = std::get_if<2>(&result))
+            std::rethrow_exception(*ep);
+        return std::get<1>(std::move(result));
+    }
+    // NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
+
+private:
+    // Exclusively-owned coroutine handle. Null after move or default
+    // construction. Destroyed in the destructor.
+    Handle handle_;
+};
+
+}  // namespace xrpl

include/xrpl/core/CoroTaskRunner.ipp

@@ -0,0 +1,448 @@
+#pragma once
+
+/**
+ * @file CoroTaskRunner.ipp
+ *
+ * CoroTaskRunner inline implementation.
+ *
+ * This file contains the business logic for managing C++20 coroutines
+ * on the JobQueue. It is included at the bottom of JobQueue.h.
+ *
+ * Data Flow: suspend / post / resume cycle
+ * =========================================
+ *
+ *         coroutine body                 CoroTaskRunner              JobQueue
+ *         --------------                 --------------              --------
+ *              |
+ *    co_await runner->suspend()
+ *              |
+ *              +--- await_suspend ------> onSuspend()
+ *              |                            ++nSuspend_ ------------> nSuspend_
+ *              |                          [coroutine is now suspended]
+ *              |
+ *              .    (externally or by yieldAndPost())
+ *              .
+ *              +--- (caller calls) -----> post()
+ *              |                            ++runCount_
+ *              |                            addJob(resume) ----------> job enqueued
+ *              |                                                        |
+ *              |                                                      [worker picks up]
+ *              |                                                        |
+ *              +--- <----- resume() <-----------------------------------+
+ *              |             --nSuspend_ ------> nSuspend_
+ *              |             swap in LocalValues (lvs_)
+ *              |             task_.handle().resume()
+ *              |                |
+ *              |    [coroutine body continues here]
+ *              |                |
+ *              |             swap out LocalValues
+ *              |             --runCount_
+ *              |             cv_.notify_all()
+ *              v
+ *
+ * Thread Safety
+ * =============
+ * - mutex_     : guards task_.handle().resume() so that post()-before-suspend
+ *                races cannot resume the coroutine while it is still running.
+ *                (See the race condition discussion in JobQueue.h)
+ * - mutexRun_ : guards runCount_ counter; used by join() to wait until
+ *                all in-flight resume operations complete.
+ * - jq_.mutex_: guards nSuspend_ increments/decrements.
+ *
+ * Common Mistakes When Modifying This File
+ * =========================================
+ *
+ * 1. Changing lock ordering.
+ *    resume() acquires locks sequentially (never held simultaneously):
+ *    jq_.mutex_ (released immediately), then mutex_ (held across resume),
+ *    then mutexRun_ (released after decrement). post() acquires only
+ *    mutexRun_. Any new code path must follow the same order.
+ *
+ * 2. Removing the shared_from_this() capture in post().
+ *    The lambda passed to addJob captures [this, sp = shared_from_this()].
+ *    If you remove sp, 'this' can be destroyed before the job runs,
+ *    causing use-after-free. The sp capture is load-bearing.
+ *
+ * 3. Forgetting to decrement nSuspend_ on a new code path.
+ *    Every ++nSuspend_ must have a matching --nSuspend_. If you add a new
+ *    suspension path (e.g. a new awaiter) and forget to decrement on resume
+ *    or on failure, JobQueue::stop() will hang.
+ *
+ * 4. Calling task_.handle().resume() without holding mutex_.
+ *    This allows a race where the coroutine runs on two threads
+ *    simultaneously. Always hold mutex_ around resume().
+ *
+ * 5. Swapping LocalValues outside of the mutex_ critical section.
+ *    The swap-in and swap-out of LocalValues must bracket the resume()
+ *    call. If you move the swap-out before the lock_guard(mutex_) is
+ *    released, you break LocalValue isolation for any code that runs
+ *    after the coroutine suspends but before the lock is dropped.
+ */
+
+namespace xrpl {
+
+/**
+ * Construct a CoroTaskRunner. Sets runCount_ to 0; does not
+ * create the coroutine. Call init() afterwards.
+ *
+ * @param jq   The JobQueue this coroutine will run on
+ * @param type Job type for scheduling priority
+ * @param name Human-readable name for logging
+ */
+inline JobQueue::CoroTaskRunner::CoroTaskRunner(
+    CreateT,
+    JobQueue& jq,
+    JobType type,
+    std::string name)
+    : jq_(jq), type_(type), name_(std::move(name))
+{
+}
+
+/**
+ * Initialize with a coroutine-returning callable.
+ * Stores the callable on the heap (FuncStore) so it outlives the
+ * coroutine frame. Coroutine frames store a reference to the
+ * callable's implicit object parameter (the lambda). If the callable
+ * is a temporary, that reference dangles after the caller returns.
+ * Keeping the callable alive here ensures the coroutine's captures
+ * remain valid.
+ *
+ * @param f Callable: CoroTask<void>(shared_ptr<CoroTaskRunner>)
+ */
+template <class F>
+void
+JobQueue::CoroTaskRunner::init(F&& f)
+{
+    using Fn = std::decay_t<F>;
+    auto store = std::make_unique<FuncStore<Fn>>(std::forward<F>(f));
+    task_ = store->func(shared_from_this());
+    storedFunc_ = std::move(store);
+}
+
+/**
+ * Destructor. Waits for any in-flight resume() to complete, then
+ * asserts (debug) that the coroutine has finished or
+ * expectEarlyExit() was called.
+ *
+ * The join() call is necessary because with async dispatch the
+ * coroutine runs on a worker thread. The gate signal (which wakes
+ * the test thread) can arrive before resume() has set finished_.
+ * join() synchronizes via mutexRun_, establishing a happens-before
+ * edge: finished_ = true -> unlock(mutexRun_) in resume() ->
+ * lock(mutexRun_) in join() -> read finished_.
+ */
+inline JobQueue::CoroTaskRunner::~CoroTaskRunner()
+{
+#ifndef NDEBUG
+    join();
+    XRPL_ASSERT(finished_, "xrpl::JobQueue::CoroTaskRunner::~CoroTaskRunner : is finished");
+#endif
+}
+
+/**
+ * Increment the JobQueue's suspended-coroutine count (nSuspend_).
+ */
+inline void
+JobQueue::CoroTaskRunner::onSuspend()
+{
+    std::scoped_lock const lock(jq_.mutex_);
+    ++jq_.nSuspend_;
+}
+
+/**
+ * Decrement nSuspend_ without resuming.
+ */
+inline void
+JobQueue::CoroTaskRunner::onUndoSuspend()
+{
+    std::scoped_lock const lock(jq_.mutex_);
+    --jq_.nSuspend_;
+}
+
+/**
+ * Return a SuspendAwaiter whose await_suspend() increments nSuspend_
+ * before the coroutine actually suspends. The caller must later call
+ * post() or resume() to continue execution.
+ *
+ * @return Awaiter for use with `co_await runner->suspend()`
+ */
+inline auto
+JobQueue::CoroTaskRunner::suspend()
+{
+    /**
+     * Custom awaiter for suspend(). Always suspends (await_ready
+     * returns false) and increments nSuspend_ in await_suspend().
+     */
+    // The C++ coroutine protocol mandates these awaiter names and
+    // instance-callable methods, which conflict with the project
+    // naming/static conventions.
+    // NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
+    struct SuspendAwaiter
+    {
+        CoroTaskRunner& runner_;  // The runner that owns this coroutine.
+
+        /**
+         * Always returns false so the coroutine suspends.
+         */
+        [[nodiscard]] bool
+        await_ready() const noexcept
+        {
+            return false;
+        }
+
+        /**
+         * Called when the coroutine suspends. Increments nSuspend_
+         * so the JobQueue knows a coroutine is waiting.
+         */
+        void
+        await_suspend(std::coroutine_handle<>) const
+        {
+            runner_.onSuspend();
+        }
+
+        void
+        await_resume() const noexcept
+        {
+        }
+    };
+    // NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
+    return SuspendAwaiter{*this};
+}
+
+/**
+ * Suspend and immediately repost on the JobQueue. Equivalent to
+ * `co_await JobQueueAwaiter{runner}` but uses an inline struct
+ * to work around a GCC-12 codegen bug (see declaration in JobQueue.h).
+ *
+ * If the JobQueue is stopping (post fails), the suspend count is
+ * undone and the coroutine continues immediately via symmetric
+ * transfer back to its own handle.
+ *
+ * @return An inline YieldPostAwaiter
+ */
+inline auto
+JobQueue::CoroTaskRunner::yieldAndPost()
+{
+    // The C++ coroutine protocol mandates these awaiter names and
+    // instance-callable methods, which conflict with the project
+    // naming/static conventions.
+    // NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
+    struct YieldPostAwaiter
+    {
+        CoroTaskRunner& runner_;
+
+        [[nodiscard]] bool
+        await_ready() const noexcept
+        {
+            return false;
+        }
+
+        /**
+         * Returns a coroutine_handle<> (symmetric transfer) rather than
+         * void + h.resume(). Two reasons:
+         *
+         *  1. h.resume() runs the coroutine nested inside this frame. A
+         *     coroutine that yields in a loop against a stopping JobQueue
+         *     fails post() every iteration, so the stack grows without
+         *     bound. Symmetric transfer is a tail call and does not nest.
+         *
+         *  2. After h.resume() returns, the coroutine may have completed
+         *     and destroyed its frame -- the frame this awaiter lives in.
+         *     Returning from await_suspend would then touch freed memory.
+         *
+         * A bool return would also avoid nesting, but GCC-12 miscompiles
+         * bool-returning await_suspend (see JobQueueAwaiter.h).
+         *
+         * @return noop_coroutine() to stay suspended (job posted);
+         *         the caller's handle to continue now (JQ stopping)
+         */
+        std::coroutine_handle<>
+        await_suspend(std::coroutine_handle<> h)
+        {
+            runner_.onSuspend();
+            if (!runner_.post())
+            {
+                runner_.onUndoSuspend();
+                return h;
+            }
+            return std::noop_coroutine();
+        }
+
+        void
+        await_resume() const noexcept
+        {
+        }
+    };
+    // NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
+    return YieldPostAwaiter{*this};
+}
+
+/**
+ * Schedule coroutine resumption as a job on the JobQueue.
+ * A shared_ptr capture (sp) prevents this CoroTaskRunner from being
+ * destroyed while the job is queued but not yet executed.
+ *
+ * @return false if the JobQueue rejected the job (shutting down)
+ */
+inline bool
+JobQueue::CoroTaskRunner::post()
+{
+    {
+        std::scoped_lock const lk(mutexRun_);
+        ++runCount_;
+    }
+
+    // sp prevents 'this' from being destroyed while the job is pending
+    if (jq_.addJob(type_, name_, [this, sp = shared_from_this()]() { resume(); }))
+    {
+        return true;
+    }
+
+    // The coroutine will not run. Undo the runCount_ increment.
+    std::scoped_lock const lk(mutexRun_);
+    --runCount_;
+    cv_.notify_all();
+    return false;
+}
+
+/**
+ * Resume the coroutine on the current thread.
+ *
+ * Steps:
+ *   1. Decrement nSuspend_ (under jq_.mutex_)
+ *   2. Swap in this coroutine's LocalValues for thread-local isolation
+ *   3. Resume the coroutine handle (under mutex_)
+ *   4. Swap out LocalValues, restoring the thread's previous state
+ *   5. Decrement runCount_ and notify join() waiters
+ *
+ * @pre post() must have been called before resume(). Direct calls
+ *      without a prior post() will corrupt runCount_ and break join().
+ * Note: runCount_ is NOT incremented here — post() already did that.
+ * This ensures join() stays blocked for the entire post->resume lifetime.
+ */
+inline void
+JobQueue::CoroTaskRunner::resume()
+{
+    {
+        std::scoped_lock const lock(jq_.mutex_);
+        --jq_.nSuspend_;
+    }
+    auto saved = detail::getLocalValues().release();
+    detail::getLocalValues().reset(&lvs_);
+    std::scoped_lock const lock(mutex_);
+    XRPL_ASSERT(
+        task_.handle() && !task_.done(),
+        "xrpl::JobQueue::CoroTaskRunner::resume : task handle is valid and not done");
+    if (task_.handle() && !task_.done())
+    {
+        task_.handle().resume();
+    }
+    else
+    {
+        // A resume() with no coroutine to run (e.g. a duplicate external
+        // post() after completion). Resuming a null or finished handle is
+        // undefined behavior, so skip it -- this matches the old
+        // Coro::resume() `if (coro_)` guard. The bookkeeping below still
+        // runs to balance the ++runCount_ done by the post() that
+        // scheduled this call.
+        JLOG(jq_.journal_.warn())
+            << "CoroTaskRunner::resume called for coroutine '" << name_
+            << "' with no runnable coroutine (duplicate post or already completed)";
+    }
+    detail::getLocalValues().release();
+    detail::getLocalValues().reset(saved);
+    if (task_.done())
+    {
+        finished_ = true;
+        // An exception that escapes a top-level coroutine body is captured
+        // by promise_type::unhandled_exception() but has no awaiter to
+        // rethrow it, so it would vanish with the frame. Surface it in the
+        // log. (The old Boost path propagated it out of resume() instead.)
+        if (auto const& ep = task_.handle().promise().exception_)
+        {
+            try
+            {
+                std::rethrow_exception(ep);
+            }
+            catch (std::exception const& e)
+            {
+                JLOG(jq_.journal_.error())
+                    << "Unhandled exception in coroutine '" << name_ << "': " << e.what();
+            }
+            catch (...)
+            {
+                JLOG(jq_.journal_.error())
+                    << "Unhandled non-standard exception in coroutine '" << name_ << "'";
+            }
+        }
+        // Break the shared_ptr cycle: frame -> shared_ptr<runner> -> this.
+        // Use std::move (not task_ = {}) so task_.handle_ is null BEFORE the
+        // frame is destroyed. operator= would destroy the frame while handle_
+        // still holds the old value -- a re-entrancy hazard on GCC-12 if
+        // frame destruction triggers runner cleanup.
+        [[maybe_unused]] auto completed = std::move(task_);
+    }
+    std::scoped_lock const lk(mutexRun_);
+    --runCount_;
+    cv_.notify_all();
+}
+
+/**
+ * @return true if the coroutine has not yet run to completion
+ */
+inline bool
+JobQueue::CoroTaskRunner::runnable() const
+{
+    // After normal completion, task_ is reset to break the shared_ptr cycle
+    // (handle_ becomes null). A null handle means the coroutine is done.
+    return task_.handle() && !task_.done();
+}
+
+/**
+ * Handle early termination when the coroutine never ran (e.g. JobQueue
+ * is stopping). Decrements nSuspend_ and destroys the coroutine frame
+ * to break the shared_ptr cycle: frame -> lambda -> runner -> frame.
+ */
+inline void
+JobQueue::CoroTaskRunner::expectEarlyExit()
+{
+    if (!finished_)
+    {
+        std::scoped_lock const lock(jq_.mutex_);
+        --jq_.nSuspend_;
+        finished_ = true;
+    }
+    // Break the shared_ptr cycle: frame -> shared_ptr<runner> -> this.
+    // The coroutine is at initial_suspend and never ran user code, so
+    // destroying it is safe. Use std::move (not task_ = {}) so
+    // task_.handle_ is null before the frame is destroyed.
+    {
+        [[maybe_unused]] auto completed = std::move(task_);
+    }
+    storedFunc_.reset();
+}
+
+/**
+ * Block until all pending/active resume operations complete.
+ * Uses cv_ + mutexRun_ to wait until runCount_ reaches 0 or
+ * finished_ becomes true. The finished_ check handles the case
+ * where resume() is called directly (without post()), which
+ * decrements runCount_ below zero. In that scenario runCount_
+ * never returns to 0, but finished_ becoming true guarantees
+ * the coroutine is done and no more resumes will occur.
+ *
+ * Note: when join() returns via the finished_ disjunct, the final
+ * resume() call may still be executing its post-completion
+ * bookkeeping (the --runCount_ / notify after finished_ is set).
+ * That is safe -- the coroutine body has fully completed and the
+ * runner is kept alive by the resume job's shared_ptr -- but
+ * callers must not assume resume() itself has returned.
+ */
+inline void
+JobQueue::CoroTaskRunner::join()
+{
+    std::unique_lock<std::mutex> lk(mutexRun_);
+    cv_.wait(lk, [this]() { return runCount_ == 0 || finished_; });
+}
+
+}  // namespace xrpl

include/xrpl/core/JobQueue.h

@@ -1,25 +1,17 @@
 #pragma once
 
 #include <xrpl/basics/LocalValue.h>
-#include <xrpl/core/ClosureCounter.h>
-#include <xrpl/core/JobTypeData.h>
-#include <xrpl/core/detail/Workers.h>
-#include <xrpl/json/json_value.h>
-
-// Include only the specific Boost.Coroutine2 headers actually used here.
-// Avoid `boost/coroutine2/all.hpp` because it transitively pulls in
-// `boost/context/pooled_fixedsize_stack.hpp`, whose `.malloc()` / `.free()`
-// member calls on `boost::pool` collide with MSVC's `_CRTDBG_MAP_ALLOC` macros
-// in Debug builds (see cmake/XrplCompiler.cmake).
 #include <xrpl/beast/insight/Collector.h>
 #include <xrpl/beast/insight/Gauge.h>
 #include <xrpl/beast/insight/Hook.h>
 #include <xrpl/beast/utility/Journal.h>
+#include <xrpl/core/ClosureCounter.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
+#include <xrpl/core/JobTypeData.h>
 #include <xrpl/core/LoadEvent.h>
-
-#include <boost/context/protected_fixedsize_stack.hpp>
-#include <boost/coroutine2/coroutine.hpp>
+#include <xrpl/core/detail/Workers.h>
+#include <xrpl/json/json_value.h>
 
 #include <atomic>
 #include <chrono>
@@ -40,10 +32,6 @@ class PerfLog;
 }  // namespace perf
 
 class Logs;
-struct CoroCreateT
-{
-    explicit CoroCreateT() = default;
-};
 
 /**
  * A pool of threads to perform work.
@@ -60,93 +48,414 @@ class JobQueue : private Workers::Callback
 {
 public:
     /**
-     * Coroutines must run to completion.
+     * C++20 coroutine lifecycle manager.
+     *
+     *  Class / Inheritance / Dependency Diagram
+     *  =========================================
+     *
+     *  std::enable_shared_from_this<CoroTaskRunner>
+     *            ^
+     *            | (public inheritance)
+     *            |
+     *  CoroTaskRunner
+     *  +---------------------------------------------------+
+     *  | - lvs_         : detail::LocalValues               |
+     *  | - jq_          : JobQueue&                         |
+     *  | - type_        : JobType                           |
+     *  | - name_        : std::string                       |
+     *  | - runCount_    : int         (in-flight resumes)    |
+     *  | - mutex_       : std::mutex      (coroutine guard) |
+     *  | - mutexRun_    : std::mutex      (join guard)      |
+     *  | - cv_          : condition_variable                 |
+     *  | - task_        : CoroTask<void>                    |
+     *  | - storedFunc_  : unique_ptr<FuncBase> (type-erased)|
+     *  +---------------------------------------------------+
+     *  | + init(F&&)         : set up coroutine callable    |
+     *  | + onSuspend()       : ++jq_.nSuspend_              |
+     *  | + onUndoSuspend()   : --jq_.nSuspend_              |
+     *  | + suspend()         : returns SuspendAwaiter       |
+     *  | + post()            : schedule resume on JobQueue  |
+     *  | + resume()          : resume coroutine on caller   |
+     *  | + runnable()        : !task_.done()                |
+     *  | + expectEarlyExit() : teardown for failed post     |
+     *  | + join()            : block until not running       |
+     *  +---------------------------------------------------+
+     *          |                          |
+     *          | owns                     | references
+     *          v                          v
+     *    CoroTask<void>             JobQueue
+     *    (coroutine frame)          (thread pool + nSuspend_)
+     *
+     *    FuncBase / FuncStore<F>    (type-erased heap storage
+     *                                for the coroutine lambda)
+     *
+     *  Coroutine Lifecycle (Control Flow)
+     *  ===================================
+     *
+     *  Caller thread              JobQueue worker thread
+     *  -------------              ----------------------
+     *  postCoroTask(f)
+     *    |
+     *    +-- reserve a jobCounter_ slot (reject if JQ shutting down)
+     *    +-- ++nSuspend_  (lazy start counts as suspended)
+     *    +-- make_shared<CoroTaskRunner>
+     *    +-- init(f)
+     *    |     +-- store lambda on heap (FuncStore)
+     *    |     +-- task_ = f(shared_from_this())
+     *    |           [coroutine created, suspended at initial_suspend]
+     *    +-- post()
+     *    |     +-- ++runCount_
+     *    |     +-- addJob(type_, [resume]{})
+     *    |                                    resume()
+     *    |                                      |
+     *    |                                      +-- --nSuspend_
+     *    |                                      +-- swap in LocalValues
+     *    |                                      +-- task_.handle().resume()
+     *    |                                      |     [coroutine body runs]
+     *    |                                      |     ...
+     *    |                                      |     co_await suspend()
+     *    |                                      |       +-- ++nSuspend_
+     *    |                                      |       [coroutine suspends]
+     *    |                                      +-- swap out LocalValues
+     *    |                                      +-- --runCount_
+     *    |                                      +-- cv_.notify_all()
+     *    |
+     *  post()  <-- called externally or by yieldAndPost()
+     *    +-- ++runCount_
+     *    +-- addJob(type_, [resume]{})
+     *                                         resume()
+     *                                           |
+     *                                           +-- [coroutine body continues]
+     *                                           +-- co_return
+     *                                           +-- --runCount_
+     *                                           +-- cv_.notify_all()
+     *  join()
+     *    +-- cv_.wait([]{runCount_ == 0})
+     *    +-- [done]
+     *
+     *  Usage Examples
+     *  ==============
+     *
+     *  1. Fire-and-forget coroutine (most common pattern):
+     *
+     *      jq.postCoroTask(JtClient, "MyWork",
+     *          [](auto runner) -> CoroTask<void> {
+     *              doSomeWork();
+     *              co_await runner->suspend();  // yield to other jobs
+     *              doMoreWork();
+     *              co_return;
+     *          });
+     *
+     *  2. Manually controlling suspend / resume (external trigger):
+     *
+     *      auto runner = jq.postCoroTask(JtClient, "ExtTrigger",
+     *          [&result](auto runner) -> CoroTask<void> {
+     *              startAsyncOperation(callback);
+     *              co_await runner->suspend();
+     *              // callback called runner->post() to get here
+     *              result = collectResult();
+     *              co_return;
+     *          });
+     *      // ... later, from the callback:
+     *      runner->post();   // reschedule the coroutine on the JobQueue
+     *
+     *  3. Using yieldAndPost() for automatic suspend + repost:
+     *
+     *      jq.postCoroTask(JtClient, "AutoRepost",
+     *          [](auto runner) -> CoroTask<void> {
+     *              step1();
+     *              co_await runner->yieldAndPost();  // yield + auto-repost
+     *              step2();
+     *              co_await runner->yieldAndPost();
+     *              step3();
+     *              co_return;
+     *          });
+     *
+     *  4. Checking shutdown after co_await (cooperative cancellation):
+     *
+     *      jq.postCoroTask(JtClient, "Cancellable",
+     *          [&jq](auto runner) -> CoroTask<void> {
+     *              while (moreWork()) {
+     *                  co_await runner->yieldAndPost();
+     *                  if (jq.isStopping())
+     *                      co_return;  // bail out cleanly
+     *                  processNextItem();
+     *              }
+     *              co_return;
+     *          });
+     *
+     *  Caveats / Pitfalls
+     *  ==================
+     *
+     *  BUG-RISK: Calling suspend() without a matching post()/resume().
+     *    After co_await runner->suspend(), the coroutine is parked and
+     *    nSuspend_ is incremented.  If nothing ever calls post() or
+     *    resume(), the coroutine is leaked and JobQueue::stop() will
+     *    hang forever waiting for nSuspend_ to reach zero.
+     *
+     *  BUG-RISK: Calling post() on an already-running coroutine.
+     *    post() schedules a resume() job.  If the coroutine has not
+     *    actually suspended yet (no co_await executed), the resume job
+     *    will try to call handle().resume() while the coroutine is still
+     *    running on another thread.  This is UB.  The mutex_ prevents
+     *    data corruption but the logic is wrong — always co_await
+     *    suspend() before calling post().  (The test incorrect_order()
+     *    shows this works only because mutex_ serializes the calls.)
+     *
+     *  BUG-RISK: Dropping the shared_ptr<CoroTaskRunner> before join().
+     *    The CoroTaskRunner destructor asserts that finished_ is true
+     *    (the coroutine completed). If you let the last shared_ptr die
+     *    while the coroutine is still running or suspended, you get an
+     *    assertion failure in debug and UB in release.  Always call
+     *    join() or expectEarlyExit() first.
+     *
+     *  BUG-RISK: Lambda captures outliving the coroutine frame.
+     *    The lambda passed to postCoroTask is heap-allocated (FuncStore)
+     *    to prevent dangling.  But objects captured by pointer still need
+     *    their own lifetime management.  If you capture a raw pointer to
+     *    a stack variable, and the stack frame exits before the coroutine
+     *    finishes, the pointer dangles.  Use shared_ptr or ensure the
+     *    pointed-to object outlives the coroutine.
+     *
+     *  BUG-RISK: Forgetting co_return in a void coroutine.
+     *    If the coroutine body falls off the end without co_return,
+     *    the compiler may silently treat it as co_return (per standard),
+     *    but some compilers warn.  Always write explicit co_return.
+     *
+     *  LIMITATION: CoroTaskRunner only supports CoroTask<void>.
+     *    The task_ member is CoroTask<void>.  To return values from
+     *    the top-level coroutine, write through a captured pointer
+     *    (as the tests demonstrate), or co_await inner CoroTask<T>
+     *    coroutines that return values.
+     *
+     *  LIMITATION: One coroutine per CoroTaskRunner.
+     *    init() must be called exactly once.  You cannot reuse a
+     *    CoroTaskRunner to run a second coroutine.  Create a new one
+     *    via postCoroTask() instead.
+     *
+     *  LIMITATION: No timeout on join().
+     *    join() blocks indefinitely.  If the coroutine is suspended
+     *    and never posted, join() will deadlock.  Use timed waits
+     *    on the gate pattern (condition_variable + wait_for) in tests.
      */
-    class Coro : public std::enable_shared_from_this<Coro>
+    class CoroTaskRunner : public std::enable_shared_from_this<CoroTaskRunner>
     {
     private:
+        // Per-coroutine thread-local storage. Swapped in before resume()
+        // and swapped out after, so each coroutine sees its own LocalValue
+        // state regardless of which worker thread executes it.
         detail::LocalValues lvs_;
+
+        // Back-reference to the owning JobQueue. Used to post jobs,
+        // increment/decrement nSuspend_, and acquire jq_.mutex_.
         JobQueue& jq_;
+
+        // Job type passed to addJob() when posting this coroutine.
         JobType type_;
+
+        // Human-readable name for this coroutine job (for logging).
         std::string name_;
-        bool running_{false};
+
+        // Number of in-flight resume operations (pending + active).
+        // Incremented by post(), decremented when resume() finishes.
+        // Guarded by mutexRun_. join() blocks until this reaches 0.
+        //
+        // A counter (not a bool) is needed because post() can be called
+        // from within the coroutine body (e.g. via yieldAndPost()),
+        // enqueuing a second resume while the first is still running.
+        // A bool would be clobbered: R2.post() sets true, then R1's
+        // cleanup sets false — losing the fact that R2 is still pending.
+        int runCount_ = 0;
+
+        // Serializes all coroutine resume() calls, preventing concurrent
+        // execution of the coroutine body on multiple threads. Handles the
+        // race where post() enqueues a resume before the coroutine has
+        // actually suspended (post-before-suspend pattern).
         std::mutex mutex_;
+
+        // Guards runCount_. Used with cv_ for join() to wait
+        // until all pending/active resume operations complete.
         std::mutex mutexRun_;
+
+        // Notified when runCount_ reaches zero, allowing
+        // join() waiters to wake up.
         std::condition_variable cv_;
-        boost::coroutines2::coroutine<void>::push_type* yield_{};
-        boost::coroutines2::coroutine<void>::pull_type coro_;
-#ifndef NDEBUG
+
+        // The coroutine handle wrapper. Owns the coroutine frame.
+        // Set by init(). Reset to empty in resume() upon coroutine
+        // completion (to break the shared_ptr cycle) or in
+        // expectEarlyExit() on early termination.
+        CoroTask<void> task_;
+
+        /**
+         * Type-erased base for heap-stored callables.
+         * Prevents the coroutine lambda from being destroyed before
+         * the coroutine frame is done with it.
+         *
+         * @see FuncStore
+         */
+        struct FuncBase
+        {
+            virtual ~FuncBase() = default;
+        };
+
+        /**
+         * Concrete type-erased storage for a callable of type F.
+         * The coroutine frame stores a reference to the lambda's implicit
+         * object parameter. If the lambda is a temporary, that reference
+         * dangles after the call returns. FuncStore keeps it alive on
+         * the heap for the lifetime of the CoroTaskRunner.
+         */
+        template <class F>
+        struct FuncStore : FuncBase
+        {
+            F func;  // The stored callable (coroutine lambda).
+            explicit FuncStore(F&& f) : func(std::move(f))
+            {
+            }
+        };
+
+        // Heap-allocated callable storage. Set by init(), ensures the
+        // lambda outlives the coroutine frame that references it.
+        std::unique_ptr<FuncBase> storedFunc_;
+
+        // True once the coroutine has completed or expectEarlyExit() was
+        // called. Asserted in the destructor (debug) to catch leaked
+        // runners. Available in all builds to guard expectEarlyExit()
+        // against double-decrementing nSuspend_.
         bool finished_ = false;
-#endif
 
     public:
-        template <class F>
-        Coro(CoroCreateT, JobQueue&, JobType, std::string, F&&);
-
-        // Not copy-constructible or assignable
-        Coro(Coro const&) = delete;
-        Coro&
-        operator=(Coro const&) = delete;
-
-        ~Coro();
+        /**
+         * Tag type for private construction. Prevents external code
+         * from constructing CoroTaskRunner directly. Use postCoroTask().
+         */
+        struct CreateT
+        {
+            explicit CreateT() = default;
+        };
 
         /**
-         * Suspend coroutine execution.
-         *  Effects:
-         *    The coroutine's stack is saved.
-         *    The associated Job thread is released.
-         *  Note:
-         *    The associated Job function returns.
-         *    Undefined behavior if called consecutively without a corresponding
-         * post.
+         * Construct a CoroTaskRunner. Private by convention (CreateT tag).
+         *
+         * @param jq   The JobQueue this coroutine will run on
+         * @param type Job type for scheduling priority
+         * @param name Human-readable name for logging
+         */
+        CoroTaskRunner(CreateT, JobQueue&, JobType, std::string);
+
+        CoroTaskRunner(CoroTaskRunner const&) = delete;
+        CoroTaskRunner&
+        operator=(CoroTaskRunner const&) = delete;
+
+        /**
+         * Destructor. Asserts (debug) that the coroutine has finished
+         * or expectEarlyExit() was called.
+         */
+        ~CoroTaskRunner();
+
+        /**
+         * Initialize with a coroutine-returning callable.
+         * Must be called exactly once, after the object is managed by
+         * shared_ptr (because init uses shared_from_this internally).
+         * This is handled automatically by postCoroTask().
+         *
+         * @param f Callable: CoroTask<void>(shared_ptr<CoroTaskRunner>)
+         */
+        template <class F>
+        void
+        init(F&& f);
+
+        /**
+         * Increment the JobQueue's suspended-coroutine count (nSuspend_).
+         * Called when the coroutine is about to suspend. Every call
+         * must be balanced by a corresponding decrement (via resume()
+         * or onUndoSuspend()), or JobQueue::stop() will hang.
          */
         void
-        yield() const;
+        onSuspend();
 
         /**
-         * Schedule coroutine execution.
-         * Effects:
-         *   Returns immediately.
-         *   A new job is scheduled to resume the execution of the coroutine.
-         *   When the job runs, the coroutine's stack is restored and execution
-         *     continues at the beginning of coroutine function or the
-         * statement after the previous call to yield. Undefined behavior if
-         * called after the coroutine has completed with a return (as opposed to
-         * a yield()). Undefined behavior if post() or resume() called
-         * consecutively without a corresponding yield.
+         * Decrement nSuspend_ without resuming.
+         * Used to undo onSuspend() when a scheduled post() fails
+         * (e.g. JobQueue is stopping).
+         */
+        void
+        onUndoSuspend();
+
+        /**
+         * Suspend the coroutine.
+         * The awaiter's await_suspend() increments nSuspend_ before the
+         * coroutine actually suspends. The caller must later call post()
+         * or resume() to continue execution.
          *
-         * @return true if the Coro's job is added to the JobQueue.
+         * @return An awaiter for use with `co_await runner->suspend()`
+         */
+        auto
+        suspend();
+
+        /**
+         * Suspend the coroutine and immediately repost it on the
+         * JobQueue. Combines suspend() + post() atomically inside
+         * await_suspend, so there is no window where an external
+         * event could race between the two.
+         *
+         * Equivalent to JobQueueAwaiter but defined as an inline
+         * awaiter returned from a member function. This avoids a
+         * GCC-12 coroutine codegen bug where an external awaiter
+         * struct (JobQueueAwaiter) used at multiple co_await points
+         * corrupts the coroutine state machine's resume index,
+         * causing the coroutine to hang on the third resumption.
+         *
+         * @return An awaiter for use with `co_await runner->yieldAndPost()`
+         */
+        auto
+        yieldAndPost();
+
+        /**
+         * Schedule coroutine resumption as a job on the JobQueue.
+         * Captures shared_from_this() to prevent this runner from being
+         * destroyed while the job is queued.
+         *
+         * @return true if the job was accepted; false if the JobQueue
+         *         is stopping (caller must handle cleanup)
          */
         bool
         post();
 
         /**
-         * Resume coroutine execution.
-         * Effects:
-         *    The coroutine continues execution from where it last left off
-         *      using this same thread.
-         *    If the coroutine has already completed, returns immediately
-         *      (handles the documented post-before-yield race condition).
-         * Undefined behavior if resume() or post() called consecutively
-         *   without a corresponding yield.
+         * Resume the coroutine on the current thread.
+         * Decrements nSuspend_, swaps in LocalValues, resumes the
+         * coroutine handle, swaps out LocalValues, and notifies join()
+         * waiters. Lock ordering (sequential, non-overlapping):
+         * jq_.mutex_ -> mutex_ -> mutexRun_.
+         *
+         * @pre post() must have been called before resume(). Direct
+         *      calls without a prior post() will corrupt runCount_
+         *      and break join().
          */
         void
         resume();
 
         /**
-         * Returns true if the Coro is still runnable (has not returned).
+         * @return true if the coroutine has not yet run to completion
          */
-        [[nodiscard]] bool
+        bool
         runnable() const;
 
         /**
-         * Once called, the Coro allows early exit without an assert.
+         * Handle early termination when the coroutine never ran.
+         * Decrements nSuspend_ and destroys the coroutine frame to
+         * break the shared_ptr cycle (frame -> lambda -> runner -> frame).
+         * Called by postCoroTask() when post() fails.
          */
         void
         expectEarlyExit();
 
         /**
-         * Waits until coroutine returns from the user function.
+         * Block until all pending/active resume operations complete.
+         * Uses cv_ + mutexRun_ to wait until runCount_ reaches 0.
+         * Warning: deadlocks if the coroutine is suspended and never posted.
          */
         void
         join();
@@ -184,18 +493,18 @@ public:
     }
 
     /**
-     * Creates a coroutine and adds a job to the queue which will run it.
+     * Creates a C++20 coroutine and adds a job to the queue to run it.
      *
      * @param t The type of job.
      * @param name Name of the job.
-     * @param f Has a signature of void(std::shared_ptr<Coro>). Called when the
-     * job executes.
+     * @param f Callable with signature
+     * CoroTask<void>(std::shared_ptr<CoroTaskRunner>).
      *
-     * @return shared_ptr to posted Coro.  nullptr if post was not successful.
+     * @return shared_ptr to posted CoroTaskRunner. nullptr if not successful.
      */
     template <class F>
-    std::shared_ptr<Coro>
-    postCoro(JobType t, std::string const& name, F&& f);
+    std::shared_ptr<CoroTaskRunner>
+    postCoroTask(JobType t, std::string const& name, F&& f);
 
     /**
      * Jobs waiting at this priority.
@@ -256,8 +565,6 @@ public:
     isStopped() const;
 
 private:
-    friend class Coro;
-
     using JobDataMap = std::map<JobType, JobTypeData>;
 
     beast::Journal journal_;
@@ -358,88 +665,89 @@ private:
     getJobLimit(JobType type);
 };
 
-/*
-    An RPC command is received and is handled via ServerHandler(HTTP) or
-    Handler(websocket), depending on the connection type. The handler then calls
-    the JobQueue::postCoro() method to create a coroutine and run it at a later
-    point. This frees up the handler thread and allows it to continue handling
-    other requests while the RPC command completes its work asynchronously.
-
-    postCoro() creates a Coro object. When the Coro ctor is called, and its
-    coro_ member is initialized (a boost::coroutines::pull_type), execution
-    automatically passes to the coroutine, which we don't want at this point,
-    since we are still in the handler thread context. It's important to note
-   here that construction of a boost pull_type automatically passes execution to
-   the coroutine. A pull_type object automatically generates a push_type that is
-    passed as a parameter (do_yield) in the signature of the function the
-    pull_type was created with. This function is immediately called during coro_
-    construction and within it, Coro::yield_ is assigned the push_type
-    parameter (do_yield) address and called (yield()) so we can return execution
-    back to the caller's stack.
-
-    postCoro() then calls Coro::post(), which schedules a job on the job
-    queue to continue execution of the coroutine in a JobQueue worker thread at
-    some later time. When the job runs, we lock on the Coro::mutex_ and call
-    coro_ which continues where we had left off. Since we the last thing we did
-    in coro_ was call yield(), the next thing we continue with is calling the
-    function param f, that was passed into Coro ctor. It is within this
-    function body that the caller specifies what he would like to do while
-    running in the coroutine and allow them to suspend and resume execution.
-    A task that relies on other events to complete, such as path finding, calls
-    Coro::yield() to suspend its execution while waiting on those events to
-    complete and continue when signaled via the Coro::post() method.
-
-    There is a potential race condition that exists here where post() can get
-    called before yield() after f is called. Technically the problem only occurs
-    if the job that post() scheduled is executed before yield() is called.
-    If the post() job were to be executed before yield(), undefined behavior
-    would occur. The lock ensures that coro_ is not called again until we exit
-    the coroutine. At which point a scheduled resume() job waiting on the lock
-    would gain entry. resume() checks if the coroutine has already completed
-    (coro_ converts to false) and, if so, skips invoking operator() since
-    calling operator() on a completed boost::coroutine2 pull_type is undefined
-    behavior.
-
-    The race condition occurs as follows:
-
-        1- The coroutine is running.
-        2- The coroutine is about to suspend, but before it can do so, it must
-            arrange for some event to wake it up.
-        3- The coroutine arranges for some event to wake it up.
-        4- Before the coroutine can suspend, that event occurs and the
-   resumption of the coroutine is scheduled on the job queue. 5- Again, before
-   the coroutine can suspend, the resumption of the coroutine is dispatched. 6-
-   Again, before the coroutine can suspend, the resumption code runs the
-            coroutine.
-        The coroutine is now running in two threads.
-
-        The lock prevents this from happening as step 6 will block until the
-            lock is released which only happens after the coroutine completes.
-*/
-
 }  // namespace xrpl
 
-#include <xrpl/core/Coro.ipp>  // IWYU pragma: keep
+#include <xrpl/core/CoroTaskRunner.ipp>  // IWYU pragma: keep
 
 namespace xrpl {
 
+// postCoroTask — entry point for launching a C++20 coroutine on the JobQueue.
+//
+// Control Flow
+// ============
+//
+//   postCoroTask(t, name, f)
+//     |
+//     +-- 1. Reserve a jobCounter_ slot — reject if JQ shutting down
+//     |
+//     +-- 2. ++nSuspend_   (the coroutine uses lazy-start, so it is
+//     |        "suspended" from the JQ's perspective before its first resume.
+//     |        This keeps the JQ shutdown logic correct — it waits for
+//     |        nSuspend_ to reach 0).
+//     |
+//     +-- 3. Create CoroTaskRunner (shared_ptr, ref-counted)
+//     |
+//     +-- 4. runner->init(f)
+//     |        +-- Heap-allocate the lambda (FuncStore) to prevent
+//     |        |   dangling captures in the coroutine frame
+//     |        +-- task_ = f(shared_from_this())
+//     |              [coroutine created but NOT started — lazy initial_suspend]
+//     |
+//     +-- 5. runner->post()
+//     |        +-- addJob(type_, [resume]{})   → resume on worker thread
+//     |        +-- failure (JQ stopping):
+//     |              +-- runner->expectEarlyExit()
+//     |              |     --nSuspend_, destroy coroutine frame
+//     |              +-- return nullptr
+//     |
+//     +-- 6. Release the jobCounter_ slot (on return)
+//
+// Why reserve a jobCounter_ slot?
+// ===============================
+// JobQueue::stop() joins jobCounter_ before it acquires mutex_ and asserts
+// nSuspend_ == 0.  Without a reservation, stop() could run to completion in
+// the window between the ++nSuspend_ in step 2 and the balancing post() or
+// expectEarlyExit() in step 5, tripping that assert.  Holding a slot blocks
+// stop()'s join() for the whole function, closing the window.  wrap() also
+// returns nullopt once the counter is joined, so it doubles as the shutdown
+// check — a plain stopping_ read cannot do this, because the read and the
+// ++nSuspend_ are not a single atomic step.
+//
+// Why async post() instead of synchronous resume()?
+// ==================================================
+// The initial dispatch MUST use async post() so the coroutine body runs on
+// a JobQueue worker thread, not the caller's thread.  resume() swaps the
+// caller's thread-local LocalValues with the coroutine's private copy.
+// If the coroutine mutates LocalValues (e.g. thread_specific_storage test),
+// those mutations bleed back into the caller's thread-local state after the
+// swap-out, corrupting subsequent tests that share the same thread pool.
+// Async post() avoids this by running the coroutine on a worker thread whose
+// LocalValues are managed by the thread pool, not by the caller.
+//
 template <class F>
-std::shared_ptr<JobQueue::Coro>
-JobQueue::postCoro(JobType t, std::string const& name, F&& f)
+std::shared_ptr<JobQueue::CoroTaskRunner>
+JobQueue::postCoroTask(JobType t, std::string const& name, F&& f)
 {
-    /*  First param is a detail type to make construction private.
-        Last param is the function the coroutine runs. Signature of
-        void(std::shared_ptr<Coro>).
-    */
-    auto coro = std::make_shared<Coro>(CoroCreateT{}, *this, t, name, std::forward<F>(f));
-    if (!coro->post())
+    // Held until this function returns. Null once jobCounter_ is joined,
+    // which is how a shutting-down JobQueue rejects new coroutines.
+    auto const shutdownGuard = jobCounter_.wrap([]() {});
+    if (!shutdownGuard)
+        return nullptr;
+
+    // Account for the initial suspension (CoroTask uses lazy start).
     {
-        // The Coro was not successfully posted.  Disable it so it's destructor
-        // can run with no negative side effects.  Then destroy it.
-        coro->expectEarlyExit();
-        coro.reset();
+        std::scoped_lock const lock(mutex_);
+        ++nSuspend_;
     }
-    return coro;
+
+    auto runner = std::make_shared<CoroTaskRunner>(CoroTaskRunner::CreateT{}, *this, t, name);
+    runner->init(std::forward<F>(f));
+    if (!runner->post())
+    {
+        runner->expectEarlyExit();
+        runner.reset();
+    }
+    return runner;
 }
 
 }  // namespace xrpl

include/xrpl/core/JobQueueAwaiter.h

@@ -0,0 +1,212 @@
+#pragma once
+
+#include <xrpl/beast/utility/instrumentation.h>
+#include <xrpl/core/JobQueue.h>
+
+#include <coroutine>
+#include <memory>
+
+namespace xrpl {
+
+/**
+ * Awaiter that suspends and immediately reschedules on the JobQueue.
+ * Equivalent to calling yield() followed by post() in the old Coro API.
+ *
+ * Usage:
+ *     co_await JobQueueAwaiter{runner};
+ *
+ * What it waits for: The coroutine is re-queued as a job and resumes
+ * when a worker thread picks it up.
+ *
+ * Which thread resumes: A JobQueue worker thread.
+ *
+ * What await_resume() returns: void.
+ *
+ * Dependency Diagram
+ * ==================
+ *
+ *   JobQueueAwaiter
+ *   +----------------------------------------------+
+ *   | + runner : shared_ptr<CoroTaskRunner>         |
+ *   +----------------------------------------------+
+ *   | + await_ready()   -> false (always suspend)   |
+ *   | + await_suspend() -> bool (suspend or cancel) |
+ *   | + await_resume()  -> void                     |
+ *   +----------------------------------------------+
+ *          |                         |
+ *          | uses                    | uses
+ *          v                         v
+ *   CoroTaskRunner              JobQueue
+ *   .onSuspend()                (via runner->post() -> addJob)
+ *   .onUndoSuspend()
+ *   .post()
+ *
+ * Control Flow (await_suspend)
+ * ============================
+ *
+ *   co_await JobQueueAwaiter{runner}
+ *     |
+ *     +-- await_ready() -> false
+ *     +-- await_suspend(handle)
+ *           |
+ *           +-- runner->onSuspend()     // ++nSuspend_
+ *           +-- runner->post()          // addJob to JobQueue
+ *           |     |
+ *           |     +-- success? return noop_coroutine()
+ *           |     |     // coroutine stays suspended;
+ *           |     |     // worker thread will call resume()
+ *           |     +-- failure? (JQ stopping)
+ *           |           +-- runner->onUndoSuspend()  // --nSuspend_
+ *           |           +-- return handle  // symmetric transfer back
+ *           |                              // coroutine continues immediately
+ *           |                              // so it can clean up and co_return
+ *
+ * DEPRECATED — prefer `co_await runner->yieldAndPost()`
+ * =====================================================
+ *
+ * GCC-12 has a coroutine codegen bug where using this external awaiter
+ * struct at multiple co_await points in the same coroutine corrupts the
+ * state machine's resume index. After the second co_await, the third
+ * resumption enters handle().resume() but never reaches await_resume()
+ * or any subsequent user code — the coroutine hangs indefinitely.
+ *
+ * The fix is `co_await runner->yieldAndPost()`, which defines the
+ * awaiter as an inline struct inside a CoroTaskRunner member function.
+ * GCC-12 handles inline awaiters correctly at multiple co_await points.
+ *
+ * This struct is retained for single-use scenarios and documentation
+ * purposes. For any code that may use co_await in a loop or at
+ * multiple points, always use `runner->yieldAndPost()`.
+ *
+ * Usage Examples
+ * ==============
+ *
+ * 1. Yield and auto-repost (preferred — works on all compilers):
+ *
+ *     CoroTask<void> handler(auto runner) {
+ *         doPartA();
+ *         co_await runner->yieldAndPost();   // yield + repost
+ *         doPartB();                         // runs on a worker thread
+ *         co_return;
+ *     }
+ *
+ * 2. Multiple yield points in a loop:
+ *
+ *     CoroTask<void> batchProcessor(auto runner) {
+ *         for (auto& item : items) {
+ *             process(item);
+ *             co_await runner->yieldAndPost();  // let other jobs run
+ *         }
+ *         co_return;
+ *     }
+ *
+ * 3. Graceful shutdown — checking after resume:
+ *
+ *     CoroTask<void> longTask(auto runner, JobQueue& jq) {
+ *         while (hasWork()) {
+ *             co_await runner->yieldAndPost();
+ *             // If JQ is stopping, await_suspend resumes the coroutine
+ *             // immediately without re-queuing. Always check
+ *             // isStopping() to decide whether to proceed:
+ *             if (jq.isStopping())
+ *                 co_return;
+ *             doNextChunk();
+ *         }
+ *         co_return;
+ *     }
+ *
+ * Caveats / Pitfalls
+ * ==================
+ *
+ * BUG-RISK: Using a stale or null runner.
+ *   The runner shared_ptr must be valid and point to the CoroTaskRunner
+ *   that owns the coroutine currently executing. Passing a runner from
+ *   a different coroutine, or a default-constructed shared_ptr, is UB.
+ *
+ * BUG-RISK: Assuming resume happens on the same thread.
+ *   After co_await, the coroutine resumes on whatever worker thread
+ *   picks up the job. Do not rely on thread-local state unless it is
+ *   managed through LocalValue (which CoroTaskRunner automatically
+ *   swaps in/out).
+ *
+ * BUG-RISK: Ignoring the shutdown path.
+ *   When the JobQueue is stopping, post() fails and await_suspend()
+ *   resumes the coroutine immediately (symmetric transfer back to h).
+ *   The coroutine body continues on the same thread. If your code
+ *   after co_await assumes it was re-queued and is running on a worker
+ *   thread, that assumption breaks during shutdown. Always handle the
+ *   "JQ is stopping" case, either by checking jq.isStopping() or by
+ *   letting the coroutine fall through to co_return naturally.
+ *
+ * DIFFERENCE from runner->suspend() + runner->post():
+ *   Both JobQueueAwaiter and yieldAndPost() combine suspend + post
+ *   in one atomic operation. With the manual suspend()/post() pattern,
+ *   there is a window between the two calls where an external event
+ *   could race. The atomic awaiters remove that window — onSuspend()
+ *   and post() happen within the same await_suspend() call while the
+ *   coroutine is guaranteed to be suspended. Use yieldAndPost() unless
+ *   you need an external party to decide *when* to call post().
+ */
+struct JobQueueAwaiter
+{
+    // The CoroTaskRunner that owns the currently executing coroutine.
+    std::shared_ptr<JobQueue::CoroTaskRunner> runner;
+
+    /**
+     * Always returns false so the coroutine suspends.
+     */
+    // The C++ coroutine protocol mandates these awaiter names and
+    // instance-callable methods, which conflict with the project
+    // naming/static conventions.
+    // NOLINTBEGIN(readability-identifier-naming, readability-convert-member-functions-to-static)
+    [[nodiscard]] bool
+    await_ready() const noexcept
+    {
+        return false;
+    }
+
+    /**
+     * Increment nSuspend (equivalent to yield()) and schedule resume
+     * on the JobQueue (equivalent to post()). If the JobQueue is
+     * stopping, undoes the suspend count and transfers back to the
+     * coroutine so it can clean up and co_return.
+     *
+     * Returns a coroutine_handle<> (symmetric transfer) instead of
+     * bool to work around a GCC-12 codegen bug where bool-returning
+     * await_suspend leaves the coroutine in an invalid state —
+     * neither properly suspended nor resumed — causing a hang.
+     *
+     * WARNING: GCC-12 has an additional codegen bug where using this
+     * external awaiter struct at multiple co_await points in the same
+     * coroutine corrupts the state machine's resume index, causing the
+     * coroutine to hang on the third resumption. Prefer
+     * `co_await runner->yieldAndPost()` which uses an inline awaiter
+     * that GCC-12 handles correctly.
+     *
+     * @return noop_coroutine() to stay suspended (job posted);
+     *         the caller's handle to resume immediately (JQ stopping)
+     */
+    std::coroutine_handle<>
+    await_suspend(std::coroutine_handle<> h)
+    {
+        XRPL_ASSERT(runner, "xrpl::JobQueueAwaiter::await_suspend : runner is valid");
+        runner->onSuspend();
+        if (!runner->post())
+        {
+            // JobQueue is stopping. Undo the suspend count and
+            // transfer back to the coroutine so it can clean up
+            // and co_return.
+            runner->onUndoSuspend();
+            return h;
+        }
+        return std::noop_coroutine();
+    }
+
+    void
+    await_resume() const noexcept
+    {
+    }
+    // NOLINTEND(readability-identifier-naming, readability-convert-member-functions-to-static)
+};
+
+}  // namespace xrpl

src/test/app/PathMPT_test.cpp

@@ -19,6 +19,7 @@
 
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/json/json_value.h>
@@ -123,7 +124,6 @@ public:
              .ledgerMaster = app.getLedgerMaster(),
              .consumer = c,
              .role = Role::USER,
-             .coro = {},
              .infoSub = {},
              .apiVersion = rpc::kApiVersionIfUnspecified},
             {},
@@ -135,22 +135,28 @@ public:
         numSrc.reserve(rpc::tuning::kMaxSrcCur);
         for (std::uint8_t i = 0; i < rpc::tuning::kMaxSrcCur; ++i)
             numSrc.push_back(makeMptID(i, bob));
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = xrpl::test::detail::rpf(alice, bob, usd, numSrc);
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.waitFor(5s));
         BEAST_EXPECT(!result.isMember(jss::error));
 
         // Test more than rpc::tuning::max_src_cur source currencies.
         numSrc.push_back(makeMptID(rpc::tuning::kMaxSrcCur, bob));
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = xrpl::test::detail::rpf(alice, bob, usd, numSrc);
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.waitFor(5s));
         BEAST_EXPECT(result.isMember(jss::error));
@@ -162,22 +168,28 @@ public:
             auto curm = MPTTester({.env = env, .issuer = alice, .holders = {bob}});
             numSrc.push_back(curm.issuanceID());
         }
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = xrpl::test::detail::rpf(alice, bob, usd, {});
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.waitFor(5s));
         BEAST_EXPECT(!result.isMember(jss::error));
 
         // Test more than rpc::tuning::max_auto_src_cur source currencies.
         auto curm = MPTTester({.env = env, .issuer = alice, .holders = {bob}});
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = xrpl::test::detail::rpf(alice, bob, usd, {});
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.waitFor(5s));
         BEAST_EXPECT(result.isMember(jss::error));

src/test/app/Path_test.cpp

@@ -26,6 +26,7 @@
 
 #include <xrpl/basics/base_uint.h>
 #include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/json/json_reader.h>
@@ -163,7 +164,6 @@ public:
              .ledgerMaster = app.getLedgerMaster(),
              .consumer = c,
              .role = Role::USER,
-             .coro = {},
              .infoSub = {},
              .apiVersion = rpc::kApiVersionIfUnspecified},
             {},
@@ -188,11 +188,14 @@ public:
 
         json::Value result;
         Gate g;
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = std::move(params);
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
 
         using namespace std::chrono_literals;
@@ -274,7 +277,6 @@ public:
              .ledgerMaster = app.getLedgerMaster(),
              .consumer = c,
              .role = Role::USER,
-             .coro = {},
              .infoSub = {},
              .apiVersion = rpc::kApiVersionIfUnspecified},
             {},
@@ -282,21 +284,27 @@ public:
         json::Value result;
         Gate g;
         // Test rpc::tuning::max_src_cur source currencies.
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = rpf(Account("alice"), Account("bob"), rpc::tuning::kMaxSrcCur);
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.waitFor(5s));
         BEAST_EXPECT(!result.isMember(jss::error));
 
         // Test more than rpc::tuning::max_src_cur source currencies.
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = rpf(Account("alice"), Account("bob"), rpc::tuning::kMaxSrcCur + 1);
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.waitFor(5s));
         BEAST_EXPECT(result.isMember(jss::error));
@@ -304,22 +312,28 @@ public:
         // Test rpc::tuning::max_auto_src_cur source currencies.
         for (auto i = 0; i < (rpc::tuning::kMaxAutoSrcCur - 1); ++i)
             env.trust(Account("alice")[std::to_string(i + 100)](100), "bob");
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = rpf(Account("alice"), Account("bob"), 0);
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.waitFor(5s));
         BEAST_EXPECT(!result.isMember(jss::error));
 
         // Test more than rpc::tuning::max_auto_src_cur source currencies.
         env.trust(Account("alice")["AUD"](100), "bob");
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = rpf(Account("alice"), Account("bob"), 0);
-            context.coro = coro;
             rpc::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.waitFor(5s));
         BEAST_EXPECT(result.isMember(jss::error));

src/test/app/PayChan_test.cpp

@@ -24,7 +24,9 @@
 #include <xrpl/basics/strHex.h>
 #include <xrpl/beast/unit_test/suite.h>
 #include <xrpl/beast/utility/Zero.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
+#include <xrpl/core/JobQueue.h>
 #include <xrpl/core/ServiceRegistry.h>
 #include <xrpl/json/json_value.h>
 #include <xrpl/json/to_string.h>
@@ -1638,7 +1640,6 @@ struct PayChan_test : public beast::unit_test::Suite
              .ledgerMaster = app.getLedgerMaster(),
              .consumer = c,
              .role = Role::USER,
-             .coro = {},
              .infoSub = {},
              .apiVersion = rpc::kApiVersionIfUnspecified},
             {},
@@ -1654,10 +1655,13 @@ struct PayChan_test : public beast::unit_test::Suite
         BEAST_EXPECT(context.loadType == resource::kFeeReferenceRpc);
         json::Value result;
         Gate g;
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
-            context.coro = coro;
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             result = doChannelVerify(context);
             g.signal();
+            co_return;
         });
 
         using namespace std::chrono_literals;
@@ -1705,7 +1709,6 @@ struct PayChan_test : public beast::unit_test::Suite
              .ledgerMaster = app.getLedgerMaster(),
              .consumer = c,
              .role = Role::ADMIN,  // channel_authorize requires ADMIN or canSign()
-             .coro = {},
              .infoSub = {},
              .apiVersion = rpc::kApiVersionIfUnspecified},
             {},
@@ -1720,10 +1723,13 @@ struct PayChan_test : public beast::unit_test::Suite
         BEAST_EXPECT(context.loadType == resource::kFeeReferenceRpc);
         json::Value result;
         Gate g;
-        app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
-            context.coro = coro;
+        // Safe capture: the test blocks on g.waitFor() until the coroutine
+        // completes, so the captured locals outlive the coroutine.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
             result = doChannelAuthorize(context);
             g.signal();
+            co_return;
         });
 
         using namespace std::chrono_literals;

src/test/core/CoroTask_test.cpp

@@ -0,0 +1,581 @@
+#include <test/jtx/Env.h>
+#include <test/jtx/envconfig.h>
+
+#include <xrpld/core/Config.h>
+
+#include <xrpl/basics/LocalValue.h>
+#include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/core/CoroTask.h>
+#include <xrpl/core/Job.h>
+#include <xrpl/core/JobQueue.h>
+#include <xrpl/core/JobQueueAwaiter.h>
+
+#include <array>
+#include <chrono>
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+// Tests intentionally capture state in coroutine lambdas; lifetimes are
+// controlled by Gate synchronization and join() before scope exit.
+// NOLINTBEGIN(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+
+namespace xrpl::test {
+
+/**
+ * Test suite for the C++20 coroutine primitives: CoroTask, CoroTaskRunner,
+ * and JobQueueAwaiter.
+ *
+ * Dependency Diagram
+ * ==================
+ *
+ *   CoroTask_test
+ *   +-------------------------------------------------+
+ *   | + Gate (inner class) : condition_variable helper |
+ *   +-------------------------------------------------+
+ *          |  uses
+ *          v
+ *   jtx::Env  -->  JobQueue::postCoroTask()
+ *                       |
+ *                       +-- CoroTaskRunner (suspend / post / resume)
+ *                       +-- CoroTask<void> / CoroTask<T>
+ *                       +-- JobQueueAwaiter
+ *
+ * Test Coverage Matrix
+ * ====================
+ *
+ *   Test                      | Primitives exercised
+ *   --------------------------+----------------------------------------------
+ *   testVoidCompletion        | CoroTask<void> basic lifecycle
+ *   testCorrectOrder          | suspend() -> join() -> post() -> complete
+ *   testIncorrectOrder        | post() before suspend() (race-safe path)
+ *   testJobQueueAwaiter       | JobQueueAwaiter + yieldAndPost suspend/repost
+ *   testThreadSpecificStorage | LocalValue isolation across coroutines
+ *   testExceptionPropagation  | CoroTask<void> exception via co_await
+ *   testMultipleYields        | N sequential suspend/resume cycles
+ *   testValueReturn           | CoroTask<T> co_return value
+ *   testValueException        | CoroTask<T> exception via co_await
+ *   testValueChaining         | nested CoroTask<T> -> CoroTask<T>
+ *   testShutdownRejection     | postCoroTask returns nullptr when stopping
+ */
+class CoroTask_test : public beast::unit_test::Suite
+{
+public:
+    /**
+     * Simple one-shot gate for synchronizing between test thread
+     * and coroutine worker threads. signal() sets the flag;
+     * waitFor() blocks until signaled or timeout.
+     */
+    class Gate
+    {
+    private:
+        std::condition_variable cv_;
+        std::mutex mutex_;
+        bool signaled_ = false;
+
+    public:
+        /**
+         * Block until signaled or timeout expires.
+         *
+         * @param relTime Maximum duration to wait
+         *
+         * @return true if signaled before timeout
+         */
+        template <class Rep, class Period>
+        bool
+        waitFor(std::chrono::duration<Rep, Period> const& relTime)
+        {
+            std::unique_lock<std::mutex> lk(mutex_);
+            auto b = cv_.wait_for(lk, relTime, [this] { return signaled_; });
+            signaled_ = false;
+            return b;
+        }
+
+        /**
+         * Signal the gate, waking any waiting thread.
+         */
+        void
+        signal()
+        {
+            std::scoped_lock const lk(mutex_);
+            signaled_ = true;
+            cv_.notify_all();
+        }
+    };
+
+    // NOTE: All coroutine lambdas passed to postCoroTask use explicit
+    // pointer-by-value captures instead of [&] to work around a GCC 14
+    // bug where reference captures in coroutine lambdas are corrupted
+    // in the coroutine frame.
+
+    /**
+     * CoroTask<void> runs to completion and runner becomes non-runnable.
+     */
+    void
+    testVoidCompletion()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("void completion");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g;
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [gp = &g](auto) -> CoroTask<void> {
+                gp->signal();
+                co_return;
+            });
+        BEAST_EXPECT(runner);
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        runner->join();
+        BEAST_EXPECT(!runner->runnable());
+    }
+
+    /**
+     * Correct order: suspend, join, post, complete.
+     * Mirrors existing Coroutine_test::correct_order.
+     */
+    void
+    testCorrectOrder()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("correct order");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g1, g2;
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
+                g1p->signal();
+                co_await runner->suspend();
+                g2p->signal();
+                co_return;
+            });
+        BEAST_EXPECT(runner);
+        if (!BEAST_EXPECT(g1.waitFor(5s)))
+            return;
+        runner->join();
+        runner->post();
+        if (!BEAST_EXPECT(g2.waitFor(5s)))
+            return;
+        runner->join();
+    }
+
+    /**
+     * Incorrect order: post() before suspend(). Verifies the
+     * race-safe path. Mirrors Coroutine_test::incorrect_order.
+     */
+    void
+    testIncorrectOrder()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("incorrect order");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g;
+        env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [gp = &g](auto runner) -> CoroTask<void> {
+                runner->post();
+                co_await runner->suspend();
+                gp->signal();
+                co_return;
+            });
+        BEAST_EXPECT(g.waitFor(5s));
+    }
+
+    /**
+     * Suspend + auto-repost across multiple yield points, using the
+     * external JobQueueAwaiter struct for the first suspension and the
+     * inline yieldAndPost() awaiter for the second. JobQueueAwaiter is
+     * used at only one co_await point per coroutine (see the GCC-12
+     * multi-use warning in JobQueueAwaiter.h).
+     */
+    void
+    testJobQueueAwaiter()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("JobQueueAwaiter");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g;
+        std::vector<int> steps;
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [sp = &steps, gp = &g](auto runner) -> CoroTask<void> {
+                sp->push_back(1);
+                co_await JobQueueAwaiter{runner};
+                sp->push_back(2);
+                co_await runner->yieldAndPost();
+                sp->push_back(3);
+                gp->signal();
+                co_return;
+            });
+        BEAST_EXPECT(runner);
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        runner->join();
+        BEAST_EXPECT(steps == std::vector<int>({1, 2, 3}));
+    }
+
+    /**
+     * Per-coroutine LocalValue isolation. Each coroutine sees its own
+     * copy of thread-local state. Mirrors Coroutine_test::thread_specific_storage.
+     */
+    void
+    testThreadSpecificStorage()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("thread specific storage");
+        Env env(*this);
+
+        auto& jq = env.app().getJobQueue();
+
+        static constexpr int kN = 4;
+        std::array<std::shared_ptr<JobQueue::CoroTaskRunner>, kN> a;
+
+        LocalValue<int> lv(-1);
+        BEAST_EXPECT(*lv == -1);
+
+        Gate g;
+        jq.addJob(JtClient, "LocalValTest", [&]() {
+            this->BEAST_EXPECT(*lv == -1);
+            *lv = -2;
+            this->BEAST_EXPECT(*lv == -2);
+            g.signal();
+        });
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        BEAST_EXPECT(*lv == -1);
+
+        for (int i = 0; i < kN; ++i)
+        {
+            jq.postCoroTask(
+                JtClient,
+                "CoroTaskTest",
+                [this, ap = &a, gp = &g, lvp = &lv, id = i](auto runner) -> CoroTask<void> {
+                    (*ap)[id] = runner;
+                    gp->signal();
+                    co_await runner->suspend();
+
+                    this->BEAST_EXPECT(**lvp == -1);
+                    **lvp = id;
+                    this->BEAST_EXPECT(**lvp == id);
+                    gp->signal();
+                    co_await runner->suspend();
+
+                    this->BEAST_EXPECT(**lvp == id);
+                    co_return;
+                });
+            if (!BEAST_EXPECT(g.waitFor(5s)))
+                return;
+            a[i]->join();
+        }
+        for (auto const& r : a)
+        {
+            r->post();
+            if (!BEAST_EXPECT(g.waitFor(5s)))
+                return;
+            r->join();
+        }
+        for (auto const& r : a)
+        {
+            r->post();
+            r->join();
+        }
+
+        jq.addJob(JtClient, "LocalValTest", [&]() {
+            this->BEAST_EXPECT(*lv == -2);
+            g.signal();
+        });
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        BEAST_EXPECT(*lv == -1);
+    }
+
+    /**
+     * An exception thrown in an awaited CoroTask<void> is rethrown into
+     * the awaiting coroutine by await_resume(), with the original
+     * message intact. (An exception escaping the top-level body has no
+     * awaiter to rethrow it; it is captured by unhandled_exception()
+     * and logged by CoroTaskRunner::resume().)
+     */
+    void
+    testExceptionPropagation()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("exception propagation");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g;
+        std::string what;
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [wp = &what, gp = &g](auto) -> CoroTask<void> {
+                auto inner = []() -> CoroTask<void> {
+                    throw std::runtime_error("test exception");
+                    co_return;
+                };
+                try
+                {
+                    co_await inner();
+                }
+                catch (std::runtime_error const& e)
+                {
+                    *wp = e.what();
+                }
+                gp->signal();
+                co_return;
+            });
+        BEAST_EXPECT(runner);
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        runner->join();
+        BEAST_EXPECT(what == "test exception");
+        BEAST_EXPECT(!runner->runnable());
+    }
+
+    /**
+     * Multiple sequential suspend/resume cycles via co_await.
+     */
+    void
+    testMultipleYields()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("multiple yields");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g;
+        int counter = 0;
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [cp = &counter, gp = &g](auto runner) -> CoroTask<void> {
+                ++(*cp);
+                gp->signal();
+                co_await runner->suspend();
+                ++(*cp);
+                gp->signal();
+                co_await runner->suspend();
+                ++(*cp);
+                gp->signal();
+                co_return;
+            });
+        BEAST_EXPECT(runner);
+
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        BEAST_EXPECT(counter == 1);
+        runner->join();
+
+        runner->post();
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        BEAST_EXPECT(counter == 2);
+        runner->join();
+
+        runner->post();
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        BEAST_EXPECT(counter == 3);
+        runner->join();
+        BEAST_EXPECT(!runner->runnable());
+    }
+
+    /**
+     * CoroTask<T> returns a value via co_return. Outer coroutine
+     * extracts it with co_await.
+     */
+    void
+    testValueReturn()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("value return");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g;
+        int result = 0;
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [rp = &result, gp = &g](auto) -> CoroTask<void> {
+                auto inner = []() -> CoroTask<int> { co_return 42; };
+                *rp = co_await inner();
+                gp->signal();
+                co_return;
+            });
+        BEAST_EXPECT(runner);
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        runner->join();
+        BEAST_EXPECT(result == 42);
+        BEAST_EXPECT(!runner->runnable());
+    }
+
+    /**
+     * CoroTask<T> propagates exceptions from inner coroutines.
+     * Outer coroutine catches via try/catch around co_await.
+     */
+    void
+    testValueException()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("value exception");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g;
+        std::string what;
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [wp = &what, gp = &g](auto) -> CoroTask<void> {
+                auto inner = []() -> CoroTask<int> {
+                    throw std::runtime_error("inner error");
+                    co_return 0;
+                };
+                try
+                {
+                    co_await inner();
+                }
+                catch (std::runtime_error const& e)
+                {
+                    *wp = e.what();
+                }
+                gp->signal();
+                co_return;
+            });
+        BEAST_EXPECT(runner);
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        runner->join();
+        BEAST_EXPECT(what == "inner error");
+        BEAST_EXPECT(!runner->runnable());
+    }
+
+    /**
+     * CoroTask<T> chaining. Nested value-returning coroutines
+     * compose via co_await.
+     */
+    void
+    testValueChaining()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("value chaining");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        Gate g;
+        int result = 0;
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [rp = &result, gp = &g](auto) -> CoroTask<void> {
+                auto add = [](int a, int b) -> CoroTask<int> { co_return a + b; };
+                auto mul = [add](int a, int b) -> CoroTask<int> {
+                    int const sum = co_await add(a, b);
+                    co_return sum * 2;
+                };
+                *rp = co_await mul(3, 4);
+                gp->signal();
+                co_return;
+            });
+        BEAST_EXPECT(runner);
+        if (!BEAST_EXPECT(g.waitFor(5s)))
+            return;
+        runner->join();
+        BEAST_EXPECT(result == 14);  // (3 + 4) * 2
+        BEAST_EXPECT(!runner->runnable());
+    }
+
+    /**
+     * postCoroTask returns nullptr when JobQueue is stopping.
+     */
+    void
+    testShutdownRejection()
+    {
+        using namespace std::chrono_literals;
+        using namespace jtx;
+
+        testcase("shutdown rejection");
+
+        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
+            cfg->forceMultiThread = true;
+            return cfg;
+        }));
+
+        // Stop the JobQueue
+        env.app().getJobQueue().stop();
+
+        auto runner = env.app().getJobQueue().postCoroTask(
+            JtClient, "CoroTaskTest", [](auto) -> CoroTask<void> { co_return; });
+        BEAST_EXPECT(!runner);
+    }
+
+    void
+    run() override
+    {
+        testVoidCompletion();
+        testCorrectOrder();
+        testIncorrectOrder();
+        testJobQueueAwaiter();
+        testThreadSpecificStorage();
+        testExceptionPropagation();
+        testMultipleYields();
+        testValueReturn();
+        testValueException();
+        testValueChaining();
+        testShutdownRejection();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(CoroTask, core, xrpl);
+
+}  // namespace xrpl::test
+
+// NOLINTEND(cppcoreguidelines-avoid-capturing-lambda-coroutines)

src/test/core/Coroutine_test.cpp

@@ -6,6 +6,7 @@
 
 #include <xrpl/basics/LocalValue.h>
 #include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
 #include <xrpl/core/JobQueue.h>
 
@@ -49,6 +50,11 @@ public:
         }
     };
 
+    // NOTE: All coroutine lambdas passed to postCoroTask use explicit
+    // pointer-by-value captures instead of [&] to work around a GCC 14
+    // bug where reference captures in coroutine lambdas are corrupted
+    // in the coroutine frame.
+
     void
     correctOrder()
     {
@@ -63,14 +69,22 @@ public:
         }));
 
         Gate g1, g2;
-        std::shared_ptr<JobQueue::Coro> c;
-        env.app().getJobQueue().postCoro(JtClient, "CoroTest", [&](auto const& cr) {
-            c = cr;
-            g1.signal();
-            c->yield();
-            g2.signal();
-        });
-        BEAST_EXPECT(g1.waitFor(5s));
+        std::shared_ptr<JobQueue::CoroTaskRunner> c;
+        env.app().getJobQueue().postCoroTask(
+            JtClient,
+            "CoroTest",
+            // Safe capture: the test blocks on the Gates until the coroutine
+            // completes, so the captured pointers outlive the coroutine.
+            // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+            [cp = &c, g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
+                *cp = runner;
+                g1p->signal();
+                co_await runner->suspend();
+                g2p->signal();
+                co_return;
+            });
+        if (!BEAST_EXPECT(g1.waitFor(5s)))
+            return;
         c->join();
         c->post();
         BEAST_EXPECT(g2.waitFor(5s));
@@ -90,11 +104,22 @@ public:
         }));
 
         Gate g;
-        env.app().getJobQueue().postCoro(JtClient, "CoroTest", [&](auto const& c) {
-            c->post();
-            c->yield();
-            g.signal();
-        });
+        env.app().getJobQueue().postCoroTask(
+            JtClient,
+            "CoroTest",
+            // Safe capture: the test blocks on the Gate until the coroutine
+            // completes, so the captured pointer outlives the coroutine.
+            // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+            [gp = &g](auto runner) -> CoroTask<void> {
+                // Schedule a resume before suspending.  The posted job
+                // cannot actually call resume() until the current resume()
+                // releases CoroTaskRunner::mutex_, which only happens after
+                // the coroutine suspends at co_await.
+                runner->post();
+                co_await runner->suspend();
+                gp->signal();
+                co_return;
+            });
         BEAST_EXPECT(g.waitFor(5s));
     }
 
@@ -110,7 +135,7 @@ public:
         auto& jq = env.app().getJobQueue();
 
         static int const kN = 4;
-        std::array<std::shared_ptr<JobQueue::Coro>, kN> a;
+        std::array<std::shared_ptr<JobQueue::CoroTaskRunner>, kN> a;
 
         LocalValue<int> lv(-1);
         BEAST_EXPECT(*lv == -1);
@@ -127,26 +152,36 @@ public:
 
         for (int i = 0; i < kN; ++i)
         {
-            jq.postCoro(JtClient, "CoroTest", [&, id = i](auto const& c) {
-                a[id] = c;
-                g.signal();
-                c->yield();
+            jq.postCoroTask(
+                JtClient,
+                "CoroTest",
+                // Safe capture: the test drives every coroutine to completion
+                // via the Gate/post()/join() sequences below, so the captured
+                // pointers outlive the coroutines.
+                // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+                [this, ap = &a, gp = &g, lvp = &lv, id = i](auto runner) -> CoroTask<void> {
+                    (*ap)[id] = runner;
+                    gp->signal();
+                    co_await runner->suspend();
 
-                this->BEAST_EXPECT(*lv == -1);
-                *lv = id;
-                this->BEAST_EXPECT(*lv == id);
-                g.signal();
-                c->yield();
+                    this->BEAST_EXPECT(**lvp == -1);
+                    **lvp = id;
+                    this->BEAST_EXPECT(**lvp == id);
+                    gp->signal();
+                    co_await runner->suspend();
 
-                this->BEAST_EXPECT(*lv == id);
-            });
-            BEAST_EXPECT(g.waitFor(5s));
+                    this->BEAST_EXPECT(**lvp == id);
+                    co_return;
+                });
+            if (!BEAST_EXPECT(g.waitFor(5s)))
+                return;
             a[i]->join();
         }
         for (auto const& c : a)
         {
             c->post();
-            BEAST_EXPECT(g.waitFor(5s));
+            if (!BEAST_EXPECT(g.waitFor(5s)))
+                return;
             c->join();
         }
         for (auto const& c : a)

src/test/core/JobQueue_test.cpp

@@ -1,11 +1,11 @@
 #include <test/jtx/Env.h>
 
 #include <xrpl/beast/unit_test/suite.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
 #include <xrpl/core/JobQueue.h>
 
 #include <atomic>
-#include <memory>
 
 namespace xrpl::test {
 
@@ -46,87 +46,114 @@ class JobQueue_test : public beast::unit_test::Suite
         }
     }
 
+    // NOTE: All coroutine lambdas passed to postCoroTask use explicit
+    // pointer-by-value captures instead of [&] to work around a GCC 14
+    // bug where reference captures in coroutine lambdas are corrupted
+    // in the coroutine frame.
+
     void
-    testPostCoro()
+    testPostCoroTask()
     {
         jtx::Env env{*this};
 
         JobQueue& jQueue = env.app().getJobQueue();
         {
-            // Test repeated post()s until the Coro completes.
+            // Test repeated post()s until the coroutine completes.
             std::atomic<int> yieldCount{0};
-            auto const coro = jQueue.postCoro(
+            auto const runner = jQueue.postCoroTask(
                 JtClient,
                 "PostCoroTest1",
-                [&yieldCount](std::shared_ptr<JobQueue::Coro> const& coroCopy) {
-                    while (++yieldCount < 4)
-                        coroCopy->yield();
+                // Safe capture: the test blocks below until the coroutine
+                // completes, so the captured pointer outlives the coroutine.
+                // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+                [ycp = &yieldCount](auto runner) -> CoroTask<void> {
+                    while (++(*ycp) < 4)
+                        co_await runner->suspend();
+                    co_return;
                 });
-            BEAST_EXPECT(coro != nullptr);
+            BEAST_EXPECT(runner != nullptr);
 
             // Wait for the Job to run and yield.
             while (yieldCount == 0)
                 ;
 
-            // Now re-post until the Coro says it is done.
+            // Now re-post until the CoroTaskRunner says it is done.
             int old = yieldCount;
-            while (coro->runnable())
+            while (runner->runnable())
             {
-                BEAST_EXPECT(coro->post());
+                BEAST_EXPECT(runner->post());
                 while (old == yieldCount)
                 {
                 }
-                coro->join();
+                runner->join();
                 BEAST_EXPECT(++old == yieldCount);
             }
             BEAST_EXPECT(yieldCount == 4);
         }
         {
-            // Test repeated resume()s until the Coro completes.
+            // Test repeated post()+join()s until the coroutine completes.
             int yieldCount{0};
-            auto const coro = jQueue.postCoro(
+            auto const runner = jQueue.postCoroTask(
                 JtClient,
                 "PostCoroTest2",
-                [&yieldCount](std::shared_ptr<JobQueue::Coro> const& coroCopy) {
-                    while (++yieldCount < 4)
-                        coroCopy->yield();
+                // Safe capture: the test blocks below until the coroutine
+                // completes, so the captured pointer outlives the coroutine.
+                // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+                [ycp = &yieldCount](auto runner) -> CoroTask<void> {
+                    while (++(*ycp) < 4)
+                        co_await runner->suspend();
+                    co_return;
                 });
-            if (!coro)
+            if (!runner)
             {
-                // There's no good reason we should not get a Coro, but we
+                // There's no good reason we should not get a runner, but we
                 // can't continue without one.
                 BEAST_EXPECT(false);
                 return;
             }
 
             // Wait for the Job to run and yield.
-            coro->join();
+            runner->join();
 
-            // Now resume until the Coro says it is done.
+            // Now post()+join() until the CoroTaskRunner says it is done.
+            // resume() requires a prior post() (see the precondition on
+            // CoroTaskRunner::resume()), so the posted job performs the
+            // resume and join() blocks until it completes. yieldCount is
+            // deliberately not atomic: the mutexRun_ handoff inside join()
+            // must provide the happens-before edge that makes the
+            // increment visible to this thread.
             int old = yieldCount;
-            while (coro->runnable())
+            while (runner->runnable())
             {
-                coro->resume();  // Resume runs synchronously on this thread.
+                BEAST_EXPECT(runner->post());
+                runner->join();
                 BEAST_EXPECT(++old == yieldCount);
             }
             BEAST_EXPECT(yieldCount == 4);
         }
         {
             // If the JobQueue is stopped, we should no
-            // longer be able to add a Coro (and calling postCoro() should
-            // return false).
+            // longer be able to post a coroutine (and calling postCoroTask()
+            // should return nullptr).
             using namespace std::chrono_literals;
             jQueue.stop();
 
-            // The Coro should never run, so having the Coro access this
+            // The coroutine should never run, so having it access this
             // unprotected variable on the stack should be completely safe.
             // Not recommended for the faint of heart...
             bool unprotected = false;
-            auto const coro = jQueue.postCoro(
-                JtClient, "PostCoroTest3", [&unprotected](std::shared_ptr<JobQueue::Coro> const&) {
-                    unprotected = false;
+            auto const runner = jQueue.postCoroTask(
+                JtClient,
+                "PostCoroTest3",
+                // Safe capture: the JobQueue is stopped, so the coroutine is
+                // never started and the captured pointer is never dereferenced.
+                // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+                [up = &unprotected](auto) -> CoroTask<void> {
+                    *up = true;
+                    co_return;
                 });
-            BEAST_EXPECT(coro == nullptr);
+            BEAST_EXPECT(runner == nullptr);
+            BEAST_EXPECT(unprotected == false);
         }
     }
 
@@ -135,7 +162,7 @@ public:
     run() override
     {
         testAddJob();
-        testPostCoro();
+        testPostCoroTask();
     }
 };
 

src/test/jtx/impl/TestHelpers.cpp

@@ -23,7 +23,9 @@
 #include <xrpl/basics/strHex.h>
 #include <xrpl/beast/unit_test/suite.h>
 #include <xrpl/beast/utility/instrumentation.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
+#include <xrpl/core/JobQueue.h>
 #include <xrpl/core/ServiceRegistry.h>
 #include <xrpl/json/json_value.h>
 #include <xrpl/json/to_string.h>
@@ -223,7 +225,6 @@ findPathsRequest(
          .ledgerMaster = app.getLedgerMaster(),
          .consumer = c,
          .role = Role::USER,
-         .coro = {},
          .infoSub = {},
          .apiVersion = rpc::kApiVersionIfUnspecified},
         {},
@@ -250,11 +251,14 @@ findPathsRequest(
 
     json::Value result;
     Gate g;
-    app.getJobQueue().postCoro(JtClient, "RPC-Client", [&](auto const& coro) {
+    // Safe capture: the caller blocks on g.waitFor() until the coroutine
+    // completes, so the captured locals outlive the coroutine.
+    // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+    app.getJobQueue().postCoroTask(JtClient, "RPC-Client", [&](auto) -> CoroTask<void> {
         context.params = std::move(params);
-        context.coro = coro;
         rpc::doCommand(context, result);
         g.signal();
+        co_return;
     });
 
     using namespace std::chrono_literals;

src/xrpld/app/main/Application.cpp

@@ -1483,7 +1483,6 @@ ApplicationImp::setup(boost::program_options::variables_map const& cmdline)
              .ledgerMaster = getLedgerMaster(),
              .consumer = c,
              .role = Role::ADMIN,
-             .coro = {},
              .infoSub = {},
              .apiVersion = rpc::kApiMaximumSupportedVersion},
             jvCommand};

src/xrpld/app/main/GRPCServer.cpp

@@ -17,6 +17,7 @@
 #include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/config/BasicConfig.h>
 #include <xrpl/config/Constants.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/protocol/ErrorCodes.h>
@@ -147,13 +148,19 @@ GRPCServerImpl::CallData<Request, Response>::process()
     // ensures that finished is always true when this CallData object
     // is returned as a tag in handleRpcs(), after sending the response
     finished_ = true;
-    auto coro = app_.getJobQueue().postCoro(
-        JobType::JtRpc, "gRPC-Client", [thisShared](std::shared_ptr<JobQueue::Coro> coro) {
-            thisShared->process(coro);
+    auto runner = app_.getJobQueue().postCoroTask(
+        JobType::JtRpc,
+        "gRPC-Client",
+        // Safe capture: postCoroTask heap-allocates the lambda (FuncStore)
+        // and thisShared keeps the CallData alive until the coroutine ends.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        [thisShared](auto) -> CoroTask<void> {
+            thisShared->processRequest();
+            co_return;
         });
 
-    // If coro is null, then the JobQueue has already been shutdown
-    if (!coro)
+    // If runner is null, then the JobQueue has already been shutdown
+    if (!runner)
     {
         grpc::Status const status{grpc::StatusCode::INTERNAL, "Job Queue is already stopped"};
         responder_.FinishWithError(status, this);
@@ -162,7 +169,7 @@ GRPCServerImpl::CallData<Request, Response>::process()
 
 template <class Request, class Response>
 void
-GRPCServerImpl::CallData<Request, Response>::process(std::shared_ptr<JobQueue::Coro> coro)
+GRPCServerImpl::CallData<Request, Response>::processRequest()
 {
     try
     {
@@ -204,7 +211,6 @@ GRPCServerImpl::CallData<Request, Response>::process(std::shared_ptr<JobQueue::C
                  app_.getLedgerMaster(),
                  usage,
                  role,
-                 coro,
                  InfoSub::pointer(),
                  kApiVersion},
                 request_};

src/xrpld/app/main/GRPCServer.h

@@ -6,7 +6,6 @@
 #include <xrpld/rpc/detail/Handler.h>
 
 #include <xrpl/beast/utility/Journal.h>
-#include <xrpl/core/JobQueue.h>
 #include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 #include <xrpl/resource/Charge.h>
 #include <xrpl/resource/Consumer.h>
@@ -228,9 +227,12 @@ private:
         clone() override;
 
     private:
-        // process the request. Called inside the coroutine passed to JobQueue
+        /**
+         * Process the gRPC request. Called inside the CoroTask lambda
+         * posted to the JobQueue by process().
+         */
         void
-        process(std::shared_ptr<JobQueue::Coro> coro);
+        processRequest();
 
         // return load type of this RPC
         resource::Charge

src/xrpld/rpc/Context.h

@@ -3,13 +3,11 @@
 #include <xrpld/rpc/Role.h>
 
 #include <xrpl/beast/utility/Journal.h>
-#include <xrpl/core/JobQueue.h>
 #include <xrpl/json/json_value.h>
 #include <xrpl/resource/Charge.h>
 #include <xrpl/resource/Consumer.h>
 #include <xrpl/server/InfoSub.h>
 
-#include <memory>
 #include <string_view>
 
 namespace xrpl {
@@ -32,7 +30,6 @@ struct Context
     LedgerMaster& ledgerMaster;
     resource::Consumer& consumer;
     Role role;
-    std::shared_ptr<JobQueue::Coro> coro;
     InfoSub::pointer infoSub;
     unsigned int apiVersion;
 };

src/xrpld/rpc/ServerHandler.h

@@ -184,13 +184,10 @@ public:
 
 private:
     json::Value
-    processSession(
-        std::shared_ptr<WSSession> const& session,
-        std::shared_ptr<JobQueue::Coro> const& coro,
-        json::Value const& jv);
+    processSession(std::shared_ptr<WSSession> const& session, json::Value const& jv);
 
     void
-    processSession(std::shared_ptr<Session> const&, std::shared_ptr<JobQueue::Coro> coro);
+    processSession(std::shared_ptr<Session> const&);
 
     void
     processRequest(
@@ -198,7 +195,6 @@ private:
         std::string const& request,
         beast::ip::Endpoint const& remoteIPAddress,
         Output const&,
-        std::shared_ptr<JobQueue::Coro> coro,
         std::string_view forwardedFor,
         std::string_view user);
 

src/xrpld/rpc/detail/PathRequestManager.cpp

@@ -6,6 +6,7 @@
 #include <xrpld/rpc/detail/PathRequest.h>
 
 #include <xrpl/basics/Log.h>
+#include <xrpl/basics/scope.h>
 #include <xrpl/core/Job.h>
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/json/json_value.h>
@@ -137,9 +138,12 @@ PathRequestManager::updateAll(std::shared_ptr<ReadView const> const& inLedger)
                     }
                     else if (request->hasCompletion())
                     {
-                        // One-shot request with completion function
+                        // One-shot request with completion function. Fire the
+                        // completion even if doUpdate throws, so the RPC
+                        // handler blocked in doRipplePathFind is released
+                        // immediately instead of waiting out its timeout.
+                        ScopeExit const complete{[&request] { request->updateComplete(); }};
                         request->doUpdate(cache, false);
-                        request->updateComplete();
                         ++processed;
                     }
                 }

src/xrpld/rpc/detail/ServerHandler.cpp

@@ -17,6 +17,7 @@
 #include <xrpl/beast/rfc2616.h>
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/config/Constants.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/Job.h>
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/json/Output.h>
@@ -317,9 +318,23 @@ ServerHandler::onRequest(Session& session)
     }
 
     std::shared_ptr<Session> const detachedSession = session.detach();
-    auto const postResult = jobQueue_.postCoro(
-        JtClientRpc, "RPC-Client", [this, detachedSession](std::shared_ptr<JobQueue::Coro> coro) {
-            processSession(detachedSession, coro);
+    auto const postResult = jobQueue_.postCoroTask(
+        JtClientRpc,
+        "RPC-Client",
+        // Safe capture: postCoroTask heap-allocates the lambda (FuncStore),
+        // detachedSession is captured by value, and this (the ServerHandler)
+        // outlives the JobQueue jobs.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        [this, detachedSession](auto) -> CoroTask<void> {
+            try
+            {
+                processSession(detachedSession);
+            }
+            catch (std::exception const& e)
+            {
+                JLOG(journal_.error()) << "RPC-Client coroutine exception: " << e.what();
+            }
+            co_return;
         });
     if (postResult == nullptr)
     {
@@ -355,17 +370,30 @@ ServerHandler::onWSMessage(
 
     JLOG(journal_.trace()) << "Websocket received '" << jv << "'";
 
-    auto const postResult = jobQueue_.postCoro(
+    auto const postResult = jobQueue_.postCoroTask(
         JtClientWebsocket,
         "WS-Client",
-        [this, session, jv = std::move(jv)](std::shared_ptr<JobQueue::Coro> const& coro) {
-            auto const jr = this->processSession(session, coro, jv);
-            auto const s = to_string(jr);
-            auto const n = s.length();
-            boost::beast::multi_buffer sb(n);
-            sb.commit(boost::asio::buffer_copy(sb.prepare(n), boost::asio::buffer(s.c_str(), n)));
-            session->send(std::make_shared<StreambufWSMsg<decltype(sb)>>(std::move(sb)));
-            session->complete();
+        // Safe capture: postCoroTask heap-allocates the lambda (FuncStore),
+        // session and jv are captured by value, and this (the ServerHandler)
+        // outlives the JobQueue jobs.
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        [this, session, jv = std::move(jv)](auto) -> CoroTask<void> {
+            try
+            {
+                auto const jr = this->processSession(session, jv);
+                auto const s = to_string(jr);
+                auto const n = s.length();
+                boost::beast::multi_buffer sb(n);
+                sb.commit(
+                    boost::asio::buffer_copy(sb.prepare(n), boost::asio::buffer(s.c_str(), n)));
+                session->send(std::make_shared<StreambufWSMsg<decltype(sb)>>(std::move(sb)));
+                session->complete();
+            }
+            catch (std::exception const& e)
+            {
+                JLOG(journal_.error()) << "WS-Client coroutine exception: " << e.what();
+            }
+            co_return;
         });
     if (postResult == nullptr)
     {
@@ -410,10 +438,7 @@ logDuration(json::Value const& request, T const& duration, beast::Journal& journ
 }
 
 json::Value
-ServerHandler::processSession(
-    std::shared_ptr<WSSession> const& session,
-    std::shared_ptr<JobQueue::Coro> const& coro,
-    json::Value const& jv)
+ServerHandler::processSession(std::shared_ptr<WSSession> const& session, json::Value const& jv)
 {
     auto is = std::static_pointer_cast<WSInfoSub>(session->appDefined);
     if (is->getConsumer().disconnect(journal_))
@@ -480,7 +505,6 @@ ServerHandler::processSession(
                  .ledgerMaster = app_.getLedgerMaster(),
                  .consumer = is->getConsumer(),
                  .role = role,
-                 .coro = coro,
                  .infoSub = is,
                  .apiVersion = apiVersion},
                 jv,
@@ -551,18 +575,14 @@ ServerHandler::processSession(
     return jr;
 }
 
-// Run as a coroutine.
 void
-ServerHandler::processSession(
-    std::shared_ptr<Session> const& session,
-    std::shared_ptr<JobQueue::Coro> coro)
+ServerHandler::processSession(std::shared_ptr<Session> const& session)
 {
     processRequest(
         session->port(),
         buffersToString(session->request().body().data()),
         session->remoteAddress().atPort(0),
         makeOutput(*session),
-        coro,
         forwardedFor(session->request()),
         [&] -> std::string_view {
             auto const iter = session->request().find("X-User");
@@ -603,7 +623,6 @@ ServerHandler::processRequest(
     std::string const& request,
     beast::ip::Endpoint const& remoteIPAddress,
     Output const& output,
-    std::shared_ptr<JobQueue::Coro> coro,
     std::string_view forwardedFor,
     std::string_view user)
 {
@@ -863,7 +882,6 @@ ServerHandler::processRequest(
              .ledgerMaster = app_.getLedgerMaster(),
              .consumer = usage,
              .role = role,
-             .coro = coro,
              .infoSub = InfoSub::pointer(),
              .apiVersion = apiVersion},
             params,

src/xrpld/rpc/detail/Tuning.h

@@ -65,6 +65,12 @@ static constexpr int kDefaultAutoFillFeeMultiplier = 10;
 static constexpr int kDefaultAutoFillFeeDivisor = 1;
 static constexpr int kMaxPathfindsInProgress = 2;
 static constexpr int kMaxPathfindJobCount = 50;
+
+/**
+ * How long a blocking ripple_path_find call waits for the path-finding
+ * continuation before giving up and returning an internal error.
+ */
+constexpr auto kPathfindCompletionTimeout = std::chrono::seconds{30};
 static constexpr int kMaxJobQueueClients = 500;
 constexpr auto kMaxValidatedLedgerAge = std::chrono::minutes{2};
 static constexpr int kMaxRequestSize = 1000000;

src/xrpld/rpc/handlers/orderbook/RipplePathFind.cpp

@@ -7,14 +7,15 @@
 #include <xrpld/rpc/detail/RPCLedgerHelpers.h>
 #include <xrpld/rpc/detail/Tuning.h>
 
-#include <xrpl/core/JobQueue.h>
 #include <xrpl/json/json_value.h>
 #include <xrpl/protocol/ErrorCodes.h>
 #include <xrpl/protocol/RPCErr.h>
 #include <xrpl/protocol/jss.h>
 #include <xrpl/resource/Fees.h>
 
+#include <condition_variable>
 #include <memory>
+#include <mutex>
 #include <utility>
 
 namespace xrpl {
@@ -47,98 +48,62 @@ doRipplePathFind(rpc::JsonContext& context)
         PathRequest::pointer request;
         lpLedger = context.ledgerMaster.getClosedLedger();
 
-        // It doesn't look like there's much odd happening here, but you should
-        // be aware this code runs in a JobQueue::Coro, which is a coroutine.
-        // And we may be flipping around between threads.  Here's an overview:
+        // The wait below parks this JobQueue worker thread until the
+        // path-finding continuation fires. The continuation is fired by a
+        // JtUpdatePf job, which itself needs a free worker to run. Bound
+        // the number of concurrently parked workers (LegacyPathFind admits
+        // at most kMaxPathfindsInProgress non-admin requests) so that
+        // concurrent ripple_path_find calls cannot occupy every worker and
+        // stall the whole JobQueue. The guard must stay in scope until the
+        // wait completes. (The old Boost.Coroutine implementation did not
+        // need this: it suspended and released the worker instead of
+        // blocking it.)
+        rpc::LegacyPathFind const lpf(isUnlimited(context.role), context.app);
+        if (!lpf.isOk())
+            return rpcError(RpcTooBusy);
+
+        // makeLegacyPathRequest enqueues a path-finding job that runs
+        // asynchronously.  We block this thread with a condition_variable
+        // until the path-finding continuation signals completion.
+        // If makeLegacyPathRequest cannot schedule the job (e.g. during
+        // shutdown), it returns an empty request and we skip the wait.
+        // Replaces the old Coro yield/resume pattern with synchronous
+        // blocking, eliminating shutdown race conditions.
         //
-        // 1. We're running doRipplePathFind() due to a call to
-        //    ripple_path_find.  doRipplePathFind() is currently running
-        //    inside of a JobQueue::Coro using a JobQueue thread.
-        //
-        // 2. doRipplePathFind's call to makeLegacyPathRequest() enqueues the
-        //    path-finding request.  That request will (probably) run at some
-        //    indeterminate future time on a (probably different) JobQueue
-        //    thread.
-        //
-        // 3. As a continuation from that path-finding JobQueue thread, the
-        //    coroutine we're currently running in (!) is posted to the
-        //    JobQueue.  Because it is a continuation, that post won't
-        //    happen until the path-finding request completes.
-        //
-        // 4. Once the continuation is enqueued, and we have reason to think
-        //    the path-finding job is likely to run, then the coroutine we're
-        //    running in yield()s.  That means it surrenders its thread in
-        //    the JobQueue.  The coroutine is suspended, but ready to run,
-        //    because it is kept resident by a shared_ptr in the
-        //    path-finding continuation.
-        //
-        // 5. If all goes well then path-finding runs on a JobQueue thread
-        //    and executes its continuation.  The continuation posts this
-        //    same coroutine (!) to the JobQueue.
-        //
-        // 6. When the JobQueue calls this coroutine, this coroutine resumes
-        //    from the line below the coro->yield() and returns the
-        //    path-finding result.
-        //
-        // With so many moving parts, what could go wrong?
-        //
-        // Just in terms of the JobQueue refusing to add jobs at shutdown
-        // there are two specific things that can go wrong.
-        //
-        // 1. The path-finding Job queued by makeLegacyPathRequest() might be
-        //    rejected (because we're shutting down).
-        //
-        //    Fortunately this problem can be addressed by looking at the
-        //    return value of makeLegacyPathRequest().  If
-        //    makeLegacyPathRequest() cannot get a thread to run the path-find
-        //    on, then it returns an empty request.
-        //
-        // 2. The path-finding job might run, but the Coro::post() might be
-        //    rejected by the JobQueue (because we're shutting down).
-        //
-        //    We handle this case by resuming (not posting) the Coro.
-        //    By resuming the Coro, we allow the Coro to run to completion
-        //    on the current thread instead of requiring that it run on a
-        //    new thread from the JobQueue.
-        //
-        // Both of these failure modes are hard to recreate in a unit test
-        // because they are so dependent on inter-thread timing.  However
-        // the failure modes can be observed by synchronously (inside the
-        // xrpld source code) shutting down the application.  The code to
-        // do so looks like this:
-        //
-        //   context.app.signalStop();
-        //   while (! context.app.getJobQueue().jobCounter().joined()) { }
-        //
-        // The first line starts the process of shutting down the app.
-        // The second line waits until no more jobs can be added to the
-        // JobQueue before letting the thread continue.
-        //
-        // May 2017
+        // The state is shared with the continuation so that it stays alive
+        // even if we stop waiting before the continuation runs.
+        struct PathDone
+        {
+            std::mutex mtx;
+            std::condition_variable cv;
+            bool done = false;
+        };
+        auto const state = std::make_shared<PathDone>();
+
         jvResult = context.app.getPathRequestManager().makeLegacyPathRequest(
             request,
-            [&context]() {
-                // Copying the shared_ptr keeps the coroutine alive up
-                // through the return.  Otherwise the storage under the
-                // captured reference could evaporate when we return from
-                // coroCopy->resume().  This is not strictly necessary, but
-                // will make maintenance easier.
-                std::shared_ptr<JobQueue::Coro> const coroCopy{context.coro};
-                if (!coroCopy->post())
+            [state]() {
                 {
-                    // The post() failed, so we won't get a thread to let
-                    // the Coro finish.  We'll call Coro::resume() so the
-                    // Coro can finish on our thread.  Otherwise the
-                    // application will hang on shutdown.
-                    coroCopy->resume();
+                    std::scoped_lock const lk(state->mtx);
+                    state->done = true;
                 }
+                state->cv.notify_one();
             },
             context.consumer,
             lpLedger,
             context.params);
         if (request)
         {
-            context.coro->yield();
+            std::unique_lock lk(state->mtx);
+            if (!state->cv.wait_for(
+                    lk, rpc::tuning::kPathfindCompletionTimeout, [&state] { return state->done; }))
+            {
+                // Path-finding continuation never fired (e.g. shutdown
+                // race or unexpected failure). Return an internal error
+                // rather than blocking the RPC thread indefinitely.
+                return rpcError(RpcInternal);
+            }
+            lk.unlock();
             jvResult = request->doStatus(context.params);
         }