Revert "Apply transaction batches in periodic intervals (#4504)" (#4852)

This reverts commit 002893f280.

There were two files with conflicts in the automated revert:

- src/ripple/rpc/impl/RPCHelpers.h and
- src/test/rpc/JSONRPC_test.cpp

Those files were manually resolved.

Builds/CMake/RippledCore.cmake

@@ -201,7 +201,6 @@ install (
     src/ripple/basics/StringUtilities.h
     src/ripple/basics/TaggedCache.h
     src/ripple/basics/tagged_integer.h
-    src/ripple/basics/SubmitSync.h
     src/ripple/basics/ThreadSafetyAnalysis.h
     src/ripple/basics/ToString.h
     src/ripple/basics/UnorderedContainers.h

cfg/rippled-example.cfg

@@ -482,7 +482,7 @@
 #
 #   Configure the maximum number of transactions to have in the job queue
 #
-#   Must be a number between 1000 and 100000, defaults to 10000
+#   Must be a number between 100 and 1000, defaults to 250
 #
 #
 # [overlay]

cfg/rippled-reporting.cfg

@@ -454,7 +454,7 @@
 #
 #   Configure the maximum number of transactions to have in the job queue
 #
-#   Must be a number between 1000 and 100000, defaults to 10000
+#   Must be a number between 100 and 1000, defaults to 250
 #
 #
 # [overlay]

src/ripple/app/ledger/impl/LedgerMaster.cpp

@@ -549,25 +549,22 @@ void
 LedgerMaster::applyHeldTransactions()
 {
     std::lock_guard sl(m_mutex);
-    // It can be expensive to modify the open ledger even with no transactions
-    // to process. Regardless, make sure to reset held transactions with
-    // the parent.
-    if (mHeldTransactions.size())
-    {
-        app_.openLedger().modify([&](OpenView& view, beast::Journal j) {
-            bool any = false;
-            for (auto const& it : mHeldTransactions)
-            {
-                ApplyFlags flags = tapNONE;
-                auto const result =
-                    app_.getTxQ().apply(app_, view, it.second, flags, j);
-                if (result.second)
-                    any = true;
-            }
-            return any;
-        });
-    }
 
+    app_.openLedger().modify([&](OpenView& view, beast::Journal j) {
+        bool any = false;
+        for (auto const& it : mHeldTransactions)
+        {
+            ApplyFlags flags = tapNONE;
+            auto const result =
+                app_.getTxQ().apply(app_, view, it.second, flags, j);
+            if (result.second)
+                any = true;
+        }
+        return any;
+    });
+
+    // VFALCO TODO recreate the CanonicalTxSet object instead of resetting
+    // it.
     // VFALCO NOTE The hash for an open ledger is undefined so we use
     // something that is a reasonable substitute.
     mHeldTransactions.reset(app_.openLedger().current()->info().parentHash);

src/ripple/app/main/Application.cpp

@@ -1533,7 +1533,6 @@ ApplicationImp::start(bool withTimers)
     {
         setSweepTimer();
         setEntropyTimer();
-        m_networkOPs->setBatchApplyTimer();
     }
 
     m_io_latency_sampler.start();

src/ripple/app/misc/NetworkOPs.cpp

@@ -43,7 +43,6 @@
 #include <ripple/app/reporting/ReportingETL.h>
 #include <ripple/app/tx/apply.h>
 #include <ripple/basics/PerfLog.h>
-#include <ripple/basics/SubmitSync.h>
 #include <ripple/basics/UptimeClock.h>
 #include <ripple/basics/mulDiv.h>
 #include <ripple/basics/safe_cast.h>
@@ -238,7 +237,6 @@ public:
         , heartbeatTimer_(io_svc)
         , clusterTimer_(io_svc)
         , accountHistoryTxTimer_(io_svc)
-        , batchApplyTimer_(io_svc)
         , mConsensus(
               app,
               make_FeeVote(
@@ -288,12 +286,43 @@ public:
     processTransaction(
         std::shared_ptr<Transaction>& transaction,
         bool bUnlimited,
-        RPC::SubmitSync sync,
         bool bLocal,
         FailHard failType) override;
 
-    bool
+    /**
-    transactionBatch(bool drain) override;
+     * For transactions submitted directly by a client, apply batch of
+     * transactions and wait for this transaction to complete.
+     *
+     * @param transaction Transaction object.
+     * @param bUnliimited Whether a privileged client connection submitted it.
+     * @param failType fail_hard setting from transaction submission.
+     */
+    void
+    doTransactionSync(
+        std::shared_ptr<Transaction> transaction,
+        bool bUnlimited,
+        FailHard failType);
+
+    /**
+     * For transactions not submitted by a locally connected client, fire and
+     * forget. Add to batch and trigger it to be processed if there's no batch
+     * currently being applied.
+     *
+     * @param transaction Transaction object
+     * @param bUnlimited Whether a privileged client connection submitted it.
+     * @param failType fail_hard setting from transaction submission.
+     */
+    void
+    doTransactionAsync(
+        std::shared_ptr<Transaction> transaction,
+        bool bUnlimited,
+        FailHard failtype);
+
+    /**
+     * Apply transactions in batches. Continue until none are queued.
+     */
+    void
+    transactionBatch();
 
     /**
      * Attempt to apply transactions and post-process based on the results.
@@ -567,15 +596,6 @@ public:
                     << "NetworkOPs: accountHistoryTxTimer cancel error: "
                     << ec.message();
             }
-
-            ec.clear();
-            batchApplyTimer_.cancel(ec);
-            if (ec)
-            {
-                JLOG(m_journal.error())
-                    << "NetworkOPs: batchApplyTimer cancel error: "
-                    << ec.message();
-            }
         }
         // Make sure that any waitHandlers pending in our timers are done.
         using namespace std::chrono_literals;
@@ -697,9 +717,6 @@ private:
     void
     setAccountHistoryJobTimer(SubAccountHistoryInfoWeak subInfo);
 
-    void
-    setBatchApplyTimer() override;
-
     Application& app_;
     beast::Journal m_journal;
 
@@ -718,8 +735,6 @@ private:
     boost::asio::steady_timer heartbeatTimer_;
     boost::asio::steady_timer clusterTimer_;
     boost::asio::steady_timer accountHistoryTxTimer_;
-    //! This timer is for applying transaction batches.
-    boost::asio::steady_timer batchApplyTimer_;
 
     RCLConsensus mConsensus;
 
@@ -993,42 +1008,6 @@ NetworkOPsImp::setAccountHistoryJobTimer(SubAccountHistoryInfoWeak subInfo)
         [this, subInfo]() { setAccountHistoryJobTimer(subInfo); });
 }
 
-void
-NetworkOPsImp::setBatchApplyTimer()
-{
-    using namespace std::chrono_literals;
-    // 100ms lag between batch intervals provides significant throughput gains
-    // with little increased latency. Tuning this figure further will
-    // require further testing. In general, increasing this figure will
-    // also increase theoretical throughput, but with diminishing returns.
-    auto constexpr batchInterval = 100ms;
-
-    setTimer(
-        batchApplyTimer_,
-        batchInterval,
-        [this]() {
-            {
-                std::lock_guard lock(mMutex);
-                // Only do the job if there's work to do and it's not currently
-                // being done.
-                if (mTransactions.size() &&
-                    mDispatchState == DispatchState::none)
-                {
-                    if (m_job_queue.addJob(
-                            jtBATCH, "transactionBatch", [this]() {
-                                transactionBatch(false);
-                            }))
-                    {
-                        mDispatchState = DispatchState::scheduled;
-                    }
-                    return;
-                }
-            }
-            setBatchApplyTimer();
-        },
-        [this]() { setBatchApplyTimer(); });
-}
-
 void
 NetworkOPsImp::processHeartbeatTimer()
 {
@@ -1205,8 +1184,7 @@ NetworkOPsImp::submitTransaction(std::shared_ptr<STTx const> const& iTrans)
 
     m_job_queue.addJob(jtTRANSACTION, "submitTxn", [this, tx]() {
         auto t = tx;
-        processTransaction(
+        processTransaction(t, false, false, FailHard::no);
-            t, false, RPC::SubmitSync::async, false, FailHard::no);
     });
 }
 
@@ -1214,7 +1192,6 @@ void
 NetworkOPsImp::processTransaction(
     std::shared_ptr<Transaction>& transaction,
     bool bUnlimited,
-    RPC::SubmitSync sync,
     bool bLocal,
     FailHard failType)
 {
@@ -1244,7 +1221,7 @@ NetworkOPsImp::processTransaction(
     // Not concerned with local checks at this point.
     if (validity == Validity::SigBad)
     {
-        JLOG(m_journal.trace()) << "Transaction has bad signature: " << reason;
+        JLOG(m_journal.info()) << "Transaction has bad signature: " << reason;
         transaction->setStatus(INVALID);
         transaction->setResult(temBAD_SIGNATURE);
         app_.getHashRouter().setFlags(transaction->getID(), SF_BAD);
@@ -1254,72 +1231,100 @@ NetworkOPsImp::processTransaction(
     // canonicalize can change our pointer
     app_.getMasterTransaction().canonicalize(&transaction);
 
-    std::unique_lock lock(mMutex);
+    if (bLocal)
-    if (!transaction->getApplying())
+        doTransactionSync(transaction, bUnlimited, failType);
+    else
+        doTransactionAsync(transaction, bUnlimited, failType);
+}
+
+void
+NetworkOPsImp::doTransactionAsync(
+    std::shared_ptr<Transaction> transaction,
+    bool bUnlimited,
+    FailHard failType)
+{
+    std::lock_guard lock(mMutex);
+
+    if (transaction->getApplying())
+        return;
+
+    mTransactions.push_back(
+        TransactionStatus(transaction, bUnlimited, false, failType));
+    transaction->setApplying();
+
+    if (mDispatchState == DispatchState::none)
     {
-        transaction->setApplying();
+        if (m_job_queue.addJob(
-        mTransactions.push_back(
+                jtBATCH, "transactionBatch", [this]() { transactionBatch(); }))
-            TransactionStatus(transaction, bUnlimited, bLocal, failType));
+        {
-    }
+            mDispatchState = DispatchState::scheduled;
-    switch (sync)
+        }
-    {
-        using enum RPC::SubmitSync;
-        case sync:
-            do
-            {
-                // If a batch is being processed, then wait. Otherwise,
-                // process a batch.
-                if (mDispatchState == DispatchState::running)
-                    mCond.wait(lock);
-                else
-                    apply(lock);
-            } while (transaction->getApplying());
-            break;
-
-        case async:
-            // It's conceivable for the submitted transaction to be
-            // processed and its result to be modified before being returned
-            // to the client. Make a copy of the transaction and set its
-            // status to guarantee that the client gets the terSUBMITTED
-            // result in all cases.
-            transaction = std::make_shared<Transaction>(*transaction);
-            transaction->setResult(terSUBMITTED);
-            break;
-
-        case wait:
-            mCond.wait(
-                lock, [&transaction] { return !transaction->getApplying(); });
-            break;
-
-        default:
-            assert(false);
     }
 }
 
-bool
+void
-NetworkOPsImp::transactionBatch(bool const drain)
+NetworkOPsImp::doTransactionSync(
+    std::shared_ptr<Transaction> transaction,
+    bool bUnlimited,
+    FailHard failType)
 {
-    {
+    std::unique_lock<std::mutex> lock(mMutex);
-        std::unique_lock<std::mutex> lock(mMutex);
-        if (mDispatchState == DispatchState::running || mTransactions.empty())
-            return false;
 
-        do
+    if (!transaction->getApplying())
-            apply(lock);
+    {
-        while (drain && mTransactions.size());
+        mTransactions.push_back(
+            TransactionStatus(transaction, bUnlimited, true, failType));
+        transaction->setApplying();
+    }
+
+    do
+    {
+        if (mDispatchState == DispatchState::running)
+        {
+            // A batch processing job is already running, so wait.
+            mCond.wait(lock);
+        }
+        else
+        {
+            apply(lock);
+
+            if (mTransactions.size())
+            {
+                // More transactions need to be applied, but by another job.
+                if (m_job_queue.addJob(jtBATCH, "transactionBatch", [this]() {
+                        transactionBatch();
+                    }))
+                {
+                    mDispatchState = DispatchState::scheduled;
+                }
+            }
+        }
+    } while (transaction->getApplying());
+}
+
+void
+NetworkOPsImp::transactionBatch()
+{
+    std::unique_lock<std::mutex> lock(mMutex);
+
+    if (mDispatchState == DispatchState::running)
+        return;
+
+    while (mTransactions.size())
+    {
+        apply(lock);
     }
-    setBatchApplyTimer();
-    return true;
 }
 
 void
 NetworkOPsImp::apply(std::unique_lock<std::mutex>& batchLock)
 {
-    assert(!mTransactions.empty());
-    assert(mDispatchState != DispatchState::running);
     std::vector<TransactionStatus> submit_held;
     std::vector<TransactionStatus> transactions;
     mTransactions.swap(transactions);
+    assert(!transactions.empty());
+
+    assert(mDispatchState != DispatchState::running);
     mDispatchState = DispatchState::running;
 
     batchLock.unlock();
@@ -1703,9 +1708,7 @@ NetworkOPsImp::checkLastClosedLedger(
         switchLedgers = false;
     }
     else
-    {
         networkClosed = closedLedger;
-    }
 
     if (!switchLedgers)
         return false;

src/ripple/app/misc/NetworkOPs.h

@@ -71,10 +71,6 @@ enum class OperatingMode {
     FULL = 4           //!< we have the ledger and can even validate
 };
 
-namespace RPC {
-enum class SubmitSync;
-}
-
 /** Provides server functionality for clients.
 
     Clients include backend applications, local commands, and connected
@@ -127,47 +123,22 @@ public:
     virtual void
     submitTransaction(std::shared_ptr<STTx const> const&) = 0;
 
-    /** Process a transaction.
+    /**
+     * Process transactions as they arrive from the network or which are
+     * submitted by clients. Process local transactions synchronously
      *
-     * The transaction has been submitted either from the peer network or
+     * @param transaction Transaction object
-     * from a client. For client submissions, there are 3 distinct behaviors:
-     *   1) sync (default): process transactions in a batch immediately,
-     *       and return only once the transaction has been processed.
-     *   2) async: Put transaction into the batch for the next processing
-     *       interval and return immediately.
-     *   3) wait: Put transaction into the batch for the next processing
-     *       interval and return only after it is processed.
-     *
-     * @param transaction Transaction object.
      * @param bUnlimited Whether a privileged client connection submitted it.
-     * @param sync Client submission synchronous behavior type requested.
+     * @param bLocal Client submission.
-     * @param bLocal Whether submitted by client (local) or peer.
+     * @param failType fail_hard setting from transaction submission.
-     * @param failType Whether to fail hard or not.
      */
     virtual void
     processTransaction(
         std::shared_ptr<Transaction>& transaction,
         bool bUnlimited,
-        RPC::SubmitSync sync,
         bool bLocal,
         FailHard failType) = 0;
 
-    /** Apply transactions in batches.
-     *
-     * Only a single batch unless drain is set. This is to optimize performance
-     * because there is significant overhead in applying each batch, whereas
-     * processing an individual transaction is fast.
-     *
-     * Setting the drain parameter is relevant for some transaction
-     * processing unit tests that expect all submitted transactions to
-     * be processed synchronously.
-     *
-     * @param drain Whether to process batches until none remain.
-     * @return Whether any transactions were processed.
-     */
-    virtual bool
-    transactionBatch(bool drain) = 0;
-
     //--------------------------------------------------------------------------
     //
     // Owner functions
@@ -216,8 +187,6 @@ public:
     setStandAlone() = 0;
     virtual void
     setStateTimer() = 0;
-    virtual void
-    setBatchApplyTimer() = 0;
 
     virtual void
     setNeedNetworkLedger() = 0;

src/ripple/app/tx/impl/apply.cpp

@@ -134,7 +134,7 @@ applyTransaction(
     if (retryAssured)
         flags = flags | tapRETRY;
 
-    JLOG(j.trace()) << "TXN " << txn.getTransactionID()
+    JLOG(j.debug()) << "TXN " << txn.getTransactionID()
                     << (retryAssured ? "/retry" : "/final");
 
     try
@@ -142,7 +142,7 @@ applyTransaction(
         auto const result = apply(app, view, txn, flags, j);
         if (result.second)
         {
-            JLOG(j.trace())
+            JLOG(j.debug())
                 << "Transaction applied: " << transHuman(result.first);
             return ApplyResult::Success;
         }
@@ -151,17 +151,17 @@ applyTransaction(
             isTelLocal(result.first))
         {
             // failure
-            JLOG(j.trace())
+            JLOG(j.debug())
                 << "Transaction failure: " << transHuman(result.first);
             return ApplyResult::Fail;
         }
 
-        JLOG(j.trace()) << "Transaction retry: " << transHuman(result.first);
+        JLOG(j.debug()) << "Transaction retry: " << transHuman(result.first);
         return ApplyResult::Retry;
     }
     catch (std::exception const& ex)
     {
-        JLOG(j.trace()) << "Throws: " << ex.what();
+        JLOG(j.warn()) << "Throws: " << ex.what();
         return ApplyResult::Fail;
     }
 }

src/ripple/basics/SubmitSync.h

@@ -1,41 +0,0 @@
-//------------------------------------------------------------------------------
-/*
-    This file is part of rippled: https://github.com/ripple/rippled
-    Copyright (c) 2023 Ripple Labs Inc.
-
-    Permission to use, copy, modify, and/or distribute this software for any
-    purpose  with  or without fee is hereby granted, provided that the above
-    copyright notice and this permission notice appear in all copies.
-
-    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
-    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
-    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
-    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
-    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
-    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
-    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-*/
-//==============================================================================
-
-#ifndef RIPPLE_BASICS_SUBMITSYNC_H_INCLUDED
-#define RIPPLE_BASICS_SUBMITSYNC_H_INCLUDED
-
-namespace ripple {
-namespace RPC {
-
-/**
- * Possible values for defining synchronous behavior of the transaction
- * submission API.
- *   1) sync (default): Process transactions in a batch immediately,
- *       and return only once the transaction has been processed.
- *   2) async: Put transaction into the batch for the next processing
- *       interval and return immediately.
- *   3) wait: Put transaction into the batch for the next processing
- *       interval and return only after it is processed.
- */
-enum class SubmitSync { sync, async, wait };
-
-}  // namespace RPC
-}  // namespace ripple
-
-#endif

src/ripple/core/Config.h

@@ -215,7 +215,7 @@ public:
 
     // Node storage configuration
     std::uint32_t LEDGER_HISTORY = 256;
-    std::uint32_t FETCH_DEPTH = 1'000'000'000;
+    std::uint32_t FETCH_DEPTH = 1000000000;
 
     // Tunable that adjusts various parameters, typically associated
     // with hardware parameters (RAM size and CPU cores). The default
@@ -232,11 +232,10 @@ public:
     // Enable the experimental Ledger Replay functionality
     bool LEDGER_REPLAY = false;
 
-    // Work queue limits. 10000 transactions is 2 full seconds of slowdown at
+    // Work queue limits
-    // 5000/s.
+    int MAX_TRANSACTIONS = 250;
-    int MAX_TRANSACTIONS = 10'000;
+    static constexpr int MAX_JOB_QUEUE_TX = 1000;
-    static constexpr int MAX_JOB_QUEUE_TX = 100'000;
+    static constexpr int MIN_JOB_QUEUE_TX = 100;
-    static constexpr int MIN_JOB_QUEUE_TX = 1'000;
 
     // Amendment majority time
     std::chrono::seconds AMENDMENT_MAJORITY_TIME = defaultAmendmentMajorityTime;

src/ripple/overlay/impl/PeerImp.cpp

@@ -28,7 +28,6 @@
 #include <ripple/app/misc/Transaction.h>
 #include <ripple/app/misc/ValidatorList.h>
 #include <ripple/app/tx/apply.h>
-#include <ripple/basics/SubmitSync.h>
 #include <ripple/basics/UptimeClock.h>
 #include <ripple/basics/base64.h>
 #include <ripple/basics/random.h>
@@ -40,14 +39,13 @@
 #include <ripple/overlay/impl/PeerImp.h>
 #include <ripple/overlay/impl/Tuning.h>
 #include <ripple/overlay/predicates.h>
-#include <ripple/protocol/Protocol.h>
 #include <ripple/protocol/digest.h>
 
+#include <boost/algorithm/string.hpp>
 #include <boost/algorithm/string/predicate.hpp>
 #include <boost/beast/core/ostream.hpp>
 
 #include <algorithm>
-#include <chrono>
 #include <memory>
 #include <mutex>
 #include <numeric>
@@ -3111,11 +3109,7 @@ PeerImp::checkTransaction(
 
         bool const trusted(flags & SF_TRUSTED);
         app_.getOPs().processTransaction(
-            tx,
+            tx, trusted, false, NetworkOPs::FailHard::no);
-            trusted,
-            RPC::SubmitSync::async,
-            false,
-            NetworkOPs::FailHard::no);
     }
     catch (std::exception const& ex)
     {

src/ripple/protocol/TER.h

@@ -218,7 +218,6 @@ enum TERcodes : TERUnderlyingType {
     terQUEUED,       // Transaction is being held in TxQ until fee drops
     terPRE_TICKET,   // Ticket is not yet in ledger but might be on its way
     terNO_AMM,       // AMM doesn't exist for the asset pair
-    terSUBMITTED     // Has been submitted async.
 };
 
 //------------------------------------------------------------------------------

src/ripple/protocol/impl/TER.cpp

@@ -211,7 +211,6 @@ transResults()
         MAKE_ERROR(terQUEUED,                 "Held until escalated fee drops."),
         MAKE_ERROR(terPRE_TICKET,             "Ticket is not yet in ledger."),
         MAKE_ERROR(terNO_AMM,                 "AMM doesn't exist for the asset pair."),
-        MAKE_ERROR(terSUBMITTED,              "Transaction has been submitted."),
 
         MAKE_ERROR(tesSUCCESS,                "The transaction was applied. Only final in a validated ledger."),
     };

src/ripple/protocol/jss.h

@@ -636,7 +636,6 @@ JSS(sub_index);             // in: LedgerEntry
 JSS(subcommand);            // in: PathFind
 JSS(success);               // rpc
 JSS(supported);             // out: AmendmentTableImpl
-JSS(sync_mode);             // in: Submit
 JSS(system_time_offset);    // out: NetworkOPs
 JSS(tag);                   // out: Peers
 JSS(taker);                 // in: Subscribe, BookOffers

src/ripple/rpc/handlers/Submit.cpp

@@ -21,7 +21,6 @@
 #include <ripple/app/misc/HashRouter.h>
 #include <ripple/app/misc/Transaction.h>
 #include <ripple/app/tx/apply.h>
-#include <ripple/basics/SubmitSync.h>
 #include <ripple/net/RPCErr.h>
 #include <ripple/protocol/ErrorCodes.h>
 #include <ripple/resource/Fees.h>
@@ -49,10 +48,6 @@ doSubmit(RPC::JsonContext& context)
 {
     context.loadType = Resource::feeMediumBurdenRPC;
 
-    auto const sync = RPC::getSubmitSyncMode(context.params);
-    if (!sync)
-        return sync.error();
-
     if (!context.params.isMember(jss::tx_blob))
     {
         auto const failType = getFailHard(context);
@@ -68,8 +63,7 @@ doSubmit(RPC::JsonContext& context)
             context.role,
             context.ledgerMaster.getValidatedLedgerAge(),
             context.app,
-            RPC::getProcessTxnFn(context.netOps),
+            RPC::getProcessTxnFn(context.netOps));
-            *sync);
 
         ret[jss::deprecated] =
             "Signing support in the 'submit' command has been "
@@ -138,7 +132,7 @@ doSubmit(RPC::JsonContext& context)
         auto const failType = getFailHard(context);
 
         context.netOps.processTransaction(
-            tpTrans, isUnlimited(context.role), *sync, true, failType);
+            tpTrans, isUnlimited(context.role), true, failType);
     }
     catch (std::exception& e)
     {

src/ripple/rpc/handlers/SubmitMultiSigned.cpp

@@ -18,12 +18,10 @@
 //==============================================================================
 
 #include <ripple/app/ledger/LedgerMaster.h>
-#include <ripple/basics/SubmitSync.h>
 #include <ripple/protocol/ErrorCodes.h>
 #include <ripple/protocol/Feature.h>
 #include <ripple/resource/Fees.h>
 #include <ripple/rpc/Context.h>
-#include <ripple/rpc/impl/RPCHelpers.h>
 #include <ripple/rpc/impl/TransactionSign.h>
 
 namespace ripple {
@@ -39,10 +37,6 @@ doSubmitMultiSigned(RPC::JsonContext& context)
     auto const failHard = context.params[jss::fail_hard].asBool();
     auto const failType = NetworkOPs::doFailHard(failHard);
 
-    auto const sync = RPC::getSubmitSyncMode(context.params);
-    if (!sync)
-        return sync.error();
-
     return RPC::transactionSubmitMultiSigned(
         context.params,
         context.apiVersion,
@@ -50,8 +44,7 @@ doSubmitMultiSigned(RPC::JsonContext& context)
         context.role,
         context.ledgerMaster.getValidatedLedgerAge(),
         context.app,
-        RPC::getProcessTxnFn(context.netOps),
+        RPC::getProcessTxnFn(context.netOps));
-        *sync);
 }
 
 }  // namespace ripple

src/ripple/rpc/impl/RPCHelpers.cpp

@@ -1125,26 +1125,5 @@ getLedgerByContext(RPC::JsonContext& context)
     return RPC::make_error(
         rpcNOT_READY, "findCreate failed to return an inbound ledger");
 }
-
-ripple::Expected<RPC::SubmitSync, Json::Value>
-getSubmitSyncMode(Json::Value const& params)
-{
-    using enum RPC::SubmitSync;
-    if (params.isMember(jss::sync_mode))
-    {
-        std::string const syncMode = params[jss::sync_mode].asString();
-        if (syncMode == "async")
-            return async;
-        else if (syncMode == "wait")
-            return wait;
-        else if (syncMode != "sync")
-            return Unexpected(RPC::make_error(
-                rpcINVALID_PARAMS,
-                "sync_mode parameter must be one of \"sync\", \"async\", or "
-                "\"wait\"."));
-    }
-    return sync;
-}
-
 }  // namespace RPC
 }  // namespace ripple

src/ripple/rpc/impl/RPCHelpers.h

@@ -26,8 +26,6 @@
 
 #include <ripple/app/misc/NetworkOPs.h>
 #include <ripple/app/misc/TxQ.h>
-#include <ripple/basics/Expected.h>
-#include <ripple/basics/SubmitSync.h>
 #include <ripple/protocol/SecretKey.h>
 #include <ripple/rpc/Context.h>
 #include <ripple/rpc/Status.h>
@@ -294,14 +292,6 @@ keypairForSignature(
     Json::Value const& params,
     Json::Value& error,
     unsigned int apiVersion = apiVersionIfUnspecified);
-/** Helper to parse submit_mode parameter to RPC submit.
- *
- * @param params RPC parameters
- * @return Either the mode or an error object.
- */
-ripple::Expected<RPC::SubmitSync, Json::Value>
-getSubmitSyncMode(Json::Value const& params);
-
 }  // namespace RPC
 }  // namespace ripple
 

src/ripple/rpc/impl/TransactionSign.cpp

@@ -834,8 +834,7 @@ transactionSubmit(
     Role role,
     std::chrono::seconds validatedLedgerAge,
     Application& app,
-    ProcessTransactionFn const& processTransaction,
+    ProcessTransactionFn const& processTransaction)
-    RPC::SubmitSync sync)
 {
     using namespace detail;
 
@@ -861,7 +860,8 @@ transactionSubmit(
     // Finally, submit the transaction.
     try
     {
-        processTransaction(txn.second, isUnlimited(role), sync, failType);
+        // FIXME: For performance, should use asynch interface
+        processTransaction(txn.second, isUnlimited(role), true, failType);
     }
     catch (std::exception&)
     {
@@ -1072,8 +1072,7 @@ transactionSubmitMultiSigned(
     Role role,
     std::chrono::seconds validatedLedgerAge,
     Application& app,
-    ProcessTransactionFn const& processTransaction,
+    ProcessTransactionFn const& processTransaction)
-    RPC::SubmitSync sync)
 {
     auto const& ledger = app.openLedger().current();
     auto j = app.journal("RPCHandler");
@@ -1246,7 +1245,7 @@ transactionSubmitMultiSigned(
     try
     {
         // FIXME: For performance, should use asynch interface
-        processTransaction(txn.second, isUnlimited(role), sync, failType);
+        processTransaction(txn.second, isUnlimited(role), true, failType);
     }
     catch (std::exception&)
     {

src/ripple/rpc/impl/TransactionSign.h

@@ -21,7 +21,6 @@
 #define RIPPLE_RPC_TRANSACTIONSIGN_H_INCLUDED
 
 #include <ripple/app/misc/NetworkOPs.h>
-#include <ripple/basics/SubmitSync.h>
 #include <ripple/ledger/ApplyView.h>
 #include <ripple/rpc/Role.h>
 
@@ -76,7 +75,7 @@ checkFee(
 using ProcessTransactionFn = std::function<void(
     std::shared_ptr<Transaction>& transaction,
     bool bUnlimited,
-    RPC::SubmitSync sync,
+    bool bLocal,
     NetworkOPs::FailHard failType)>;
 
 inline ProcessTransactionFn
@@ -85,10 +84,9 @@ getProcessTxnFn(NetworkOPs& netOPs)
     return [&netOPs](
                std::shared_ptr<Transaction>& transaction,
                bool bUnlimited,
-               RPC::SubmitSync sync,
+               bool bLocal,
                NetworkOPs::FailHard failType) {
-        netOPs.processTransaction(
+        netOPs.processTransaction(transaction, bUnlimited, bLocal, failType);
-            transaction, bUnlimited, sync, true, failType);
     };
 }
 
@@ -111,8 +109,7 @@ transactionSubmit(
     Role role,
     std::chrono::seconds validatedLedgerAge,
     Application& app,
-    ProcessTransactionFn const& processTransaction,
+    ProcessTransactionFn const& processTransaction);
-    RPC::SubmitSync sync);
 
 /** Returns a Json::objectValue. */
 Json::Value
@@ -133,8 +130,7 @@ transactionSubmitMultiSigned(
     Role role,
     std::chrono::seconds validatedLedgerAge,
     Application& app,
-    ProcessTransactionFn const& processTransaction,
+    ProcessTransactionFn const& processTransaction);
-    RPC::SubmitSync sync);
 
 }  // namespace RPC
 }  // namespace ripple

src/test/app/Transaction_ordering_test.cpp

@@ -15,7 +15,6 @@
 */
 //==============================================================================
 
-#include <ripple/app/misc/NetworkOPs.h>
 #include <ripple/core/JobQueue.h>
 #include <ripple/protocol/ErrorCodes.h>
 #include <test/jtx.h>
@@ -92,7 +91,6 @@ struct Transaction_ordering_test : public beast::unit_test::suite
         env(tx2, ter(terPRE_SEQ));
         BEAST_EXPECT(env.seq(alice) == aliceSequence);
         env(tx1);
-        BEAST_EXPECT(env.app().getOPs().transactionBatch(false));
         env.app().getJobQueue().rendezvous();
         BEAST_EXPECT(env.seq(alice) == aliceSequence + 2);
 
@@ -145,8 +143,6 @@ struct Transaction_ordering_test : public beast::unit_test::suite
         }
 
         env(tx[0]);
-        // Apply until no more deferred/held transactions.
-        BEAST_EXPECT(env.app().getOPs().transactionBatch(true));
         env.app().getJobQueue().rendezvous();
         BEAST_EXPECT(env.seq(alice) == aliceSequence + 5);
 

src/test/jtx/Env_test.cpp

@@ -29,7 +29,6 @@
 
 #include <boost/lexical_cast.hpp>
 #include <optional>
-#include <thread>
 #include <utility>
 
 namespace ripple {
@@ -901,95 +900,6 @@ public:
         pass();
     }
 
-    void
-    testSyncSubmit()
-    {
-        using namespace jtx;
-        Env env(*this);
-
-        auto const alice = Account{"alice"};
-        auto const n = XRP(10000);
-        env.fund(n, alice);
-        BEAST_EXPECT(env.balance(alice) == n);
-
-        // submit only
-        auto applyBlobTxn = [&env](char const* syncMode, auto&&... txnArgs) {
-            auto jt = env.jt(txnArgs...);
-            Serializer s;
-            jt.stx->add(s);
-
-            Json::Value args{Json::objectValue};
-
-            args[jss::tx_blob] = strHex(s.slice());
-            args[jss::fail_hard] = true;
-            args[jss::sync_mode] = syncMode;
-
-            return env.rpc("json", "submit", args.toStyledString());
-        };
-
-        auto jr = applyBlobTxn("sync", noop(alice));
-        BEAST_EXPECT(jr[jss::result][jss::engine_result] == "tesSUCCESS");
-
-        jr = applyBlobTxn("async", noop(alice));
-        BEAST_EXPECT(jr[jss::result][jss::engine_result] == "terSUBMITTED");
-        // Make sure it gets processed before submitting and waiting.
-        env.app().getOPs().transactionBatch(true);
-
-        auto applier = [&env]() {
-            while (!env.app().getOPs().transactionBatch(false))
-                std::this_thread::sleep_for(std::chrono::milliseconds(1));
-        };
-        auto t = std::thread(applier);
-
-        jr = applyBlobTxn("wait", noop(alice));
-        BEAST_EXPECT(jr[jss::result][jss::engine_result] == "tesSUCCESS");
-        t.join();
-
-        jr = applyBlobTxn("scott", noop(alice));
-        BEAST_EXPECT(jr[jss::result][jss::error] == "invalidParams");
-
-        // sign and submit
-        auto applyJsonTxn = [&env](
-                                char const* syncMode,
-                                std::string const secret,
-                                Json::Value const& val) {
-            Json::Value args{Json::objectValue};
-            args[jss::secret] = secret;
-            args[jss::tx_json] = val;
-            args[jss::fail_hard] = true;
-            args[jss::sync_mode] = syncMode;
-
-            return env.rpc("json", "submit", args.toStyledString());
-        };
-
-        Json::Value payment;
-        auto secret = toBase58(generateSeed("alice"));
-        payment = noop("alice");
-        payment[sfSequence.fieldName] = env.seq("alice");
-        payment[sfSetFlag.fieldName] = 0;
-        jr = applyJsonTxn("sync", secret, payment);
-        BEAST_EXPECT(jr[jss::result][jss::engine_result] == "tesSUCCESS");
-
-        payment[sfSequence.fieldName] = env.seq("alice");
-        jr = applyJsonTxn("async", secret, payment);
-        BEAST_EXPECT(jr[jss::result][jss::engine_result] == "terSUBMITTED");
-
-        env.app().getOPs().transactionBatch(true);
-        payment[sfSequence.fieldName] = env.seq("alice");
-
-        auto aSeq = env.seq("alice");
-        t = std::thread(applier);
-        jr = applyJsonTxn("wait", secret, payment);
-        BEAST_EXPECT(jr[jss::result][jss::engine_result] == "tesSUCCESS");
-        t.join();
-        // Ensure the last transaction was processed.
-        BEAST_EXPECT(env.seq("alice") == aSeq + 1);
-
-        payment[sfSequence.fieldName] = env.seq("alice");
-        jr = applyJsonTxn("scott", secret, payment);
-        BEAST_EXPECT(jr[jss::result][jss::error] == "invalidParams");
-    }
-
     void
     run() override
     {
@@ -1015,7 +925,6 @@ public:
         testSignAndSubmit();
         testFeatures();
         testExceptionalShutdown();
-        testSyncSubmit();
     }
 };
 

src/test/rpc/JSONRPC_test.cpp

@@ -19,7 +19,6 @@
 
 #include <ripple/app/misc/LoadFeeTrack.h>
 #include <ripple/app/misc/TxQ.h>
-#include <ripple/basics/SubmitSync.h>
 #include <ripple/basics/contract.h>
 #include <ripple/beast/unit_test.h>
 #include <ripple/core/ConfigSections.h>
@@ -2499,7 +2498,7 @@ public:
     fakeProcessTransaction(
         std::shared_ptr<Transaction>&,
         bool,
-        SubmitSync,
+        bool,
         NetworkOPs::FailHard)
     {
         ;
@@ -2549,8 +2548,7 @@ public:
             Role role,
             std::chrono::seconds validatedLedgerAge,
             Application& app,
-            ProcessTransactionFn const& processTransaction,
+            ProcessTransactionFn const& processTransaction);
-            RPC::SubmitSync sync);
 
         using TestStuff =
             std::tuple<signFunc, submitFunc, char const*, unsigned int>;
@@ -2605,8 +2603,7 @@ public:
                             testRole,
                             1s,
                             env.app(),
-                            processTxn,
+                            processTxn);
-                            RPC::SubmitSync::sync);
                     }
 
                     std::string errStr;

src/test/rpc/RobustTransaction_test.cpp

@@ -17,7 +17,6 @@
 */
 //==============================================================================
 
-#include <ripple/app/misc/NetworkOPs.h>
 #include <ripple/beast/unit_test.h>
 #include <ripple/core/JobQueue.h>
 #include <ripple/protocol/jss.h>
@@ -89,8 +88,7 @@ public:
             }
             BEAST_EXPECT(jv[jss::result][jss::engine_result] == "tefPAST_SEQ");
 
-            // Submit future sequence transaction -- this transaction should be
+            // Submit future sequence transaction
-            // held until the sequence gap is closed.
             payment[jss::tx_json][sfSequence.fieldName] = env.seq("alice") + 1;
             jv = wsc->invoke("submit", payment);
             if (wsc->version() == 2)
@@ -116,8 +114,6 @@ public:
             }
             BEAST_EXPECT(jv[jss::result][jss::engine_result] == "tesSUCCESS");
 
-            // Apply held transactions.
-            env.app().getOPs().transactionBatch(true);
             // Wait for the jobqueue to process everything
             env.app().getJobQueue().rendezvous();