Migrate production entry points from Boost.Coroutine to C++20 coroutines

Replace all postCoro() call sites with postCoroTask() using C++20
coroutine lambdas. The key changes are:

- Remove Context::coro field (shared_ptr<JobQueue::Coro>) from
  RPC::Context, eliminating it from all aggregate initializations
- Replace RipplePathFind's yield/post/resume pattern with a local
  std::condition_variable that blocks until path-finding completes,
  avoiding colored-function infection across the RPC call chain
- Switch ServerHandler entry points (onRequest, onWSMessage) from
  postCoro to postCoroTask with co_return lambdas
- Switch GRPCServer::CallData::process() to use postCoroTask,
  rename private handler to processRequest()
- Update Path_test and AMMTest to use postCoroTask (they set
  context.coro which no longer exists)

The old postCoro() API remains available for Coroutine_test and
JobQueue_test, which will be migrated in a subsequent commit.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pratikmankawde

src/test/app/Path_test.cpp

@@ -8,6 +8,7 @@
 #include <xrpld/rpc/detail/Tuning.h>
 
 #include <xrpl/beast/unit_test.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/json/json_reader.h>
 #include <xrpl/protocol/ApiVersion.h>
@@ -131,7 +132,6 @@ class Path_test : public beast::unit_test::suite
              c,
              Role::USER,
              {},
-             {},
              RPC::apiVersionIfUnspecified},
             {},
             {}};
@@ -155,11 +155,11 @@ class Path_test : public beast::unit_test::suite
 
         Json::Value result;
         gate g;
-        app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
+        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;
@@ -240,51 +240,50 @@ class Path_test : public beast::unit_test::suite
              c,
              Role::USER,
              {},
-             {},
              RPC::apiVersionIfUnspecified},
             {},
             {}};
         Json::Value result;
         gate g;
         // Test RPC::Tuning::max_src_cur source currencies.
-        app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
+        app.getJobQueue().postCoroTask(jtCLIENT, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = rpf(Account("alice"), Account("bob"), RPC::Tuning::max_src_cur);
-            context.coro = coro;
             RPC::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.wait_for(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) {
+        app.getJobQueue().postCoroTask(jtCLIENT, "RPC-Client", [&](auto) -> CoroTask<void> {
             context.params = rpf(Account("alice"), Account("bob"), RPC::Tuning::max_src_cur + 1);
-            context.coro = coro;
             RPC::doCommand(context, result);
             g.signal();
+            co_return;
         });
         BEAST_EXPECT(g.wait_for(5s));
         BEAST_EXPECT(result.isMember(jss::error));
 
         // Test RPC::Tuning::max_auto_src_cur source currencies.
         for (auto i = 0; i < (RPC::Tuning::max_auto_src_cur - 1); ++i)
             env.trust(Account("alice")[std::to_string(i + 100)](100), "bob");
-        app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
+        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.wait_for(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) {
+        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.wait_for(5s));
         BEAST_EXPECT(result.isMember(jss::error));

src/test/jtx/impl/AMMTest.cpp

@@ -6,6 +6,7 @@
 
 #include <xrpld/rpc/RPCHandler.h>
 
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/protocol/ApiVersion.h>
 #include <xrpl/protocol/STParsedJSON.h>
 #include <xrpl/resource/Fees.h>
@@ -193,7 +194,6 @@ AMMTest::find_paths_request(
          c,
          Role::USER,
          {},
-         {},
          RPC::apiVersionIfUnspecified},
         {},
         {}};
@@ -215,11 +215,11 @@ AMMTest::find_paths_request(
 
     Json::Value result;
     gate g;
-    app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
+    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/GRPCServer.cpp

@@ -3,6 +3,7 @@
 
 #include <xrpl/beast/core/CurrentThreadName.h>
 #include <xrpl/beast/net/IPAddressConversion.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/resource/Fees.h>
 
 namespace xrpl {
@@ -99,13 +100,14 @@ 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", [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 status{grpc::StatusCode::INTERNAL, "Job Queue is already stopped"};
         responder_.FinishWithError(status, this);
@@ -114,7 +116,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
     {
@@ -156,7 +158,6 @@ GRPCServerImpl::CallData<Request, Response>::process(std::shared_ptr<JobQueue::C
                  app_.getLedgerMaster(),
                  usage,
                  role,
-                 coro,
                  InfoSub::pointer(),
                  apiVersion},
                 request_};

src/xrpld/app/main/GRPCServer.h

@@ -208,7 +208,7 @@ class GRPCServerImpl final
     private:
         // process the request. Called inside the coroutine passed to JobQueue
         void
-        process(std::shared_ptr<JobQueue::Coro> coro);
+        processRequest();
 
         // return load type of this RPC
         Resource::Charge

src/xrpld/rpc/Context.h

@@ -3,7 +3,6 @@
 #include <xrpld/rpc/Role.h>
 
 #include <xrpl/beast/utility/Journal.h>
-#include <xrpl/core/JobQueue.h>
 #include <xrpl/server/InfoSub.h>
 
 namespace xrpl {
@@ -24,7 +23,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

@@ -171,19 +171,17 @@ class ServerHandler
     Json::Value
     processSession(
         std::shared_ptr<WSSession> const& session,
-        std::shared_ptr<JobQueue::Coro> const& coro,
         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(
         Port const& port,
         std::string const& request,
         beast::IP::Endpoint const& remoteIPAddress,
         Output&&,
-        std::shared_ptr<JobQueue::Coro> coro,
         std::string_view forwardedFor,
         std::string_view user);
 

src/xrpld/rpc/detail/ServerHandler.cpp

@@ -12,6 +12,7 @@
 #include <xrpl/basics/base64.h>
 #include <xrpl/basics/contract.h>
 #include <xrpl/basics/make_SSLContext.h>
+#include <xrpl/core/CoroTask.h>
 #include <xrpl/beast/net/IPAddressConversion.h>
 #include <xrpl/beast/rfc2616.h>
 #include <xrpl/core/JobQueue.h>
@@ -284,9 +285,10 @@ ServerHandler::onRequest(Session& session)
     }
 
     std::shared_ptr<Session> detachedSession = session.detach();
-    auto const postResult = m_jobQueue.postCoro(
-        jtCLIENT_RPC, "RPC-Client", [this, detachedSession](std::shared_ptr<JobQueue::Coro> coro) {
-            processSession(detachedSession, coro);
+    auto const postResult = m_jobQueue.postCoroTask(
+        jtCLIENT_RPC, "RPC-Client", [this, detachedSession](auto) -> CoroTask<void> {
+            processSession(detachedSession);
+            co_return;
         });
     if (postResult == nullptr)
     {
@@ -322,17 +324,18 @@ ServerHandler::onWSMessage(
 
     JLOG(m_journal.trace()) << "Websocket received '" << jv << "'";
 
-    auto const postResult = m_jobQueue.postCoro(
+    auto const postResult = m_jobQueue.postCoroTask(
         jtCLIENT_WEBSOCKET,
         "WS-Client",
-        [this, session, jv = std::move(jv)](std::shared_ptr<JobQueue::Coro> const& coro) {
-            auto const jr = this->processSession(session, coro, jv);
+        [this, session, jv = std::move(jv)](auto) -> CoroTask<void> {
+            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();
+            co_return;
         });
     if (postResult == nullptr)
     {
@@ -375,7 +378,6 @@ 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)
 {
     auto is = std::static_pointer_cast<WSInfoSub>(session->appDefined);
@@ -443,7 +445,6 @@ ServerHandler::processSession(
                  app_.getLedgerMaster(),
                  is->getConsumer(),
                  role,
-                 coro,
                  is,
                  apiVersion},
                 jv,
@@ -514,18 +515,15 @@ ServerHandler::processSession(
     return jr;
 }
 
-// Run as a coroutine.
 void
 ServerHandler::processSession(
-    std::shared_ptr<Session> const& session,
-    std::shared_ptr<JobQueue::Coro> coro)
+    std::shared_ptr<Session> const& session)
 {
     processRequest(
         session->port(),
         buffers_to_string(session->request().body().data()),
         session->remoteAddress().at_port(0),
         makeOutput(*session),
-        coro,
         forwardedFor(session->request()),
         [&] {
             auto const iter = session->request().find("X-User");
@@ -562,7 +560,6 @@ ServerHandler::processRequest(
     std::string const& request,
     beast::IP::Endpoint const& remoteIPAddress,
     Output&& output,
-    std::shared_ptr<JobQueue::Coro> coro,
     std::string_view forwardedFor,
     std::string_view user)
 {
@@ -819,7 +816,6 @@ ServerHandler::processRequest(
              app_.getLedgerMaster(),
              usage,
              role,
-             coro,
              InfoSub::pointer(),
              apiVersion},
             params,