diff --git a/README.md b/README.md index f00eb4fed..efe087d1b 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,6 @@ Below are some useful docs to learn more about Clio. **For Developers**: - [How to build Clio](./docs/build-clio.md) -- [Metrics and static analysis](./docs/metrics-and-static-analysis.md) - [Coverage report](./docs/coverage-report.md) **For Operators**: diff --git a/conanfile.py b/conanfile.py index 997dee1c4..ee1d02598 100644 --- a/conanfile.py +++ b/conanfile.py @@ -28,7 +28,7 @@ class Clio(ConanFile): 'protobuf/3.21.9', 'grpc/1.50.1', 'openssl/1.1.1u', - 'xrpl/2.4.0-b1', + 'xrpl/2.4.0-b3', 'zlib/1.3.1', 'libbacktrace/cci.20210118' ] diff --git a/docker/ci/README.md b/docker/ci/README.md index edf25b6dc..cfc256f61 100644 --- a/docker/ci/README.md +++ b/docker/ci/README.md @@ -4,12 +4,13 @@ This image contains an environment to build [Clio](https://github.com/XRPLF/clio It is used in [Clio Github Actions](https://github.com/XRPLF/clio/actions) but can also be used to compile Clio locally. The image is based on Ubuntu 20.04 and contains: -- clang 16 +- clang 16.0.6 - gcc 12.3 -- doxygen 1.10 +- doxygen 1.12 - gh 2.40 -- ccache 4.8.3 -- conan +- ccache 4.10.2 +- conan 1.62 - and some other useful tools -Conan is set up to build Clio without any additional steps. There are two preset conan profiles: `clang` and `gcc` to use corresponding compiler. +Conan is set up to build Clio without any additional steps. There are two preset conan profiles: `clang` and `gcc` to use corresponding compiler. By default conan is setup to use `gcc`. +Sanitizer builds for `ASAN`, `TSAN` and `UBSAN` are enabled via conan profiles for each of the supported compilers. These can be selected using the following pattern (all lowercase): `[compiler].[sanitizer]` (e.g. `--profile gcc.tsan`). diff --git a/docker/ci/conan/clang.asan b/docker/ci/conan/clang.asan new file mode 100644 index 000000000..0ec726451 --- /dev/null +++ b/docker/ci/conan/clang.asan @@ -0,0 +1,9 @@ +include(clang) + +[options] +boost:extra_b2_flags="cxxflags=\"-fsanitize=address\" linkflags=\"-fsanitize=address\"" +boost:without_stacktrace=True +[env] +CFLAGS="-fsanitize=address" +CXXFLAGS="-fsanitize=address" +LDFLAGS="-fsanitize=address" diff --git a/docker/ci/conan/clang.tsan b/docker/ci/conan/clang.tsan new file mode 100644 index 000000000..8f314ca44 --- /dev/null +++ b/docker/ci/conan/clang.tsan @@ -0,0 +1,9 @@ +include(clang) + +[options] +boost:extra_b2_flags="cxxflags=\"-fsanitize=thread\" linkflags=\"-fsanitize=thread\"" +boost:without_stacktrace=True +[env] +CFLAGS="-fsanitize=thread" +CXXFLAGS="-fsanitize=thread" +LDFLAGS="-fsanitize=thread" diff --git a/docker/ci/conan/clang.ubsan b/docker/ci/conan/clang.ubsan new file mode 100644 index 000000000..8f6f17d06 --- /dev/null +++ b/docker/ci/conan/clang.ubsan @@ -0,0 +1,9 @@ +include(clang) + +[options] +boost:extra_b2_flags="cxxflags=\"-fsanitize=undefined\" linkflags=\"-fsanitize=undefined\"" +boost:without_stacktrace=True +[env] +CFLAGS="-fsanitize=undefined" +CXXFLAGS="-fsanitize=undefined" +LDFLAGS="-fsanitize=undefined" diff --git a/docker/ci/conan/gcc.asan b/docker/ci/conan/gcc.asan new file mode 100644 index 000000000..65d85da46 --- /dev/null +++ b/docker/ci/conan/gcc.asan @@ -0,0 +1,9 @@ +include(gcc) + +[options] +boost:extra_b2_flags="cxxflags=\"-fsanitize=address\" linkflags=\"-fsanitize=address\"" +boost:without_stacktrace=True +[env] +CFLAGS="-fsanitize=address" +CXXFLAGS="-fsanitize=address" +LDFLAGS="-fsanitize=address" diff --git a/docker/ci/conan/gcc.tsan b/docker/ci/conan/gcc.tsan new file mode 100644 index 000000000..1f13a0715 --- /dev/null +++ b/docker/ci/conan/gcc.tsan @@ -0,0 +1,9 @@ +include(gcc) + +[options] +boost:extra_b2_flags="cxxflags=\"-fsanitize=thread\" linkflags=\"-fsanitize=thread\"" +boost:without_stacktrace=True +[env] +CFLAGS="-fsanitize=thread" +CXXFLAGS="-fsanitize=thread" +LDFLAGS="-fsanitize=thread" diff --git a/docker/ci/conan/gcc.ubsan b/docker/ci/conan/gcc.ubsan new file mode 100644 index 000000000..ddbb0257d --- /dev/null +++ b/docker/ci/conan/gcc.ubsan @@ -0,0 +1,9 @@ +include(gcc) + +[options] +boost:extra_b2_flags="cxxflags=\"-fsanitize=undefined\" linkflags=\"-fsanitize=undefined\"" +boost:without_stacktrace=True +[env] +CFLAGS="-fsanitize=undefined" +CXXFLAGS="-fsanitize=undefined" +LDFLAGS="-fsanitize=undefined" diff --git a/docker/ci/dockerfile b/docker/ci/dockerfile index 197e96c62..85d661bc2 100644 --- a/docker/ci/dockerfile +++ b/docker/ci/dockerfile @@ -98,3 +98,10 @@ RUN conan profile new clang --detect \ && conan profile update "conf.tools.build:compiler_executables={\"c\": \"/usr/bin/clang-16\", \"cpp\": \"/usr/bin/clang++-16\"}" clang RUN echo "include(gcc)" >> .conan/profiles/default + +COPY conan/gcc.asan /root/.conan/profiles +COPY conan/gcc.tsan /root/.conan/profiles +COPY conan/gcc.ubsan /root/.conan/profiles +COPY conan/clang.asan /root/.conan/profiles +COPY conan/clang.tsan /root/.conan/profiles +COPY conan/clang.ubsan /root/.conan/profiles diff --git a/docs/build-clio.md b/docs/build-clio.md index c681090f3..6351b0345 100644 --- a/docs/build-clio.md +++ b/docs/build-clio.md @@ -181,3 +181,20 @@ Sometimes, during development, you need to build against a custom version of `li 4. Build Clio as you would have before. See [Building Clio](#building-clio) for details. + +## Using `clang-tidy` for static analysis + +The minimum [clang-tidy](https://clang.llvm.org/extra/clang-tidy/) version required is 19.0. + +Clang-tidy can be run by Cmake when building the project. To achieve this, you just need to provide the option `-o lint=True` for the `conan install` command: + +```sh +conan install .. --output-folder . --build missing --settings build_type=Release -o tests=True -o lint=True +``` + +By default Cmake will try to find `clang-tidy` automatically in your system. +To force Cmake to use your desired binary, set the `CLIO_CLANG_TIDY_BIN` environment variable to the path of the `clang-tidy` binary. For example: + +```sh +export CLIO_CLANG_TIDY_BIN=/opt/homebrew/opt/llvm@19/bin/clang-tidy +``` diff --git a/docs/metrics-and-static-analysis.md b/docs/metrics-and-static-analysis.md deleted file mode 100644 index 792685650..000000000 --- a/docs/metrics-and-static-analysis.md +++ /dev/null @@ -1,30 +0,0 @@ -# Metrics and static analysis - -## Prometheus metrics collection - -Clio natively supports [Prometheus](https://prometheus.io/) metrics collection. It accepts Prometheus requests on the port configured in the `server` section of the config. - -Prometheus metrics are enabled by default, and replies to `/metrics` are compressed. To disable compression, and have human readable metrics, add `"prometheus": { "enabled": true, "compress_reply": false }` to Clio's config. - -To completely disable Prometheus metrics add `"prometheus": { "enabled": false }` to Clio's config. - -It is important to know that Clio responds to Prometheus request only if they are admin requests. If you are using the admin password feature, the same password should be provided in the Authorization header of Prometheus requests. - -You can find an example docker-compose file, with Prometheus and Grafana configs, in [examples/infrastructure](../docs/examples/infrastructure/). - -## Using `clang-tidy` for static analysis - -The minimum [clang-tidy](https://clang.llvm.org/extra/clang-tidy/) version required is 19.0. - -Clang-tidy can be run by Cmake when building the project. To achieve this, you just need to provide the option `-o lint=True` for the `conan install` command: - -```sh -conan install .. --output-folder . --build missing --settings build_type=Release -o tests=True -o lint=True -``` - -By default Cmake will try to find `clang-tidy` automatically in your system. -To force Cmake to use your desired binary, set the `CLIO_CLANG_TIDY_BIN` environment variable to the path of the `clang-tidy` binary. For example: - -```sh -export CLIO_CLANG_TIDY_BIN=/opt/homebrew/opt/llvm@19/bin/clang-tidy -``` diff --git a/docs/run-clio.md b/docs/run-clio.md index 51c5a2cac..e5309a800 100644 --- a/docs/run-clio.md +++ b/docs/run-clio.md @@ -80,3 +80,15 @@ Clio will fallback to hardcoded defaults when these values are not specified in > [!TIP] > See the [example-config.json](../docs/examples/config/example-config.json) for more details. + +## Prometheus metrics collection + +Clio natively supports [Prometheus](https://prometheus.io/) metrics collection. It accepts Prometheus requests on the port configured in the `server` section of the config. + +Prometheus metrics are enabled by default, and replies to `/metrics` are compressed. To disable compression, and have human readable metrics, add `"prometheus": { "enabled": true, "compress_reply": false }` to Clio's config. + +To completely disable Prometheus metrics add `"prometheus": { "enabled": false }` to Clio's config. + +It is important to know that Clio responds to Prometheus request only if they are admin requests. If you are using the admin password feature, the same password should be provided in the Authorization header of Prometheus requests. + +You can find an example docker-compose file, with Prometheus and Grafana configs, in [examples/infrastructure](../docs/examples/infrastructure/). diff --git a/src/app/CMakeLists.txt b/src/app/CMakeLists.txt index 0559fa49c..40cbbbc14 100644 --- a/src/app/CMakeLists.txt +++ b/src/app/CMakeLists.txt @@ -1,4 +1,4 @@ add_library(clio_app) -target_sources(clio_app PRIVATE CliArgs.cpp ClioApplication.cpp WebHandlers.cpp) +target_sources(clio_app PRIVATE CliArgs.cpp ClioApplication.cpp Stopper.cpp WebHandlers.cpp) target_link_libraries(clio_app PUBLIC clio_etl clio_etlng clio_feed clio_web clio_rpc clio_migration) diff --git a/src/app/CliArgs.cpp b/src/app/CliArgs.cpp index 0c973dee6..4673c9f90 100644 --- a/src/app/CliArgs.cpp +++ b/src/app/CliArgs.cpp @@ -47,6 +47,7 @@ CliArgs::parse(int argc, char const* argv[]) ("conf,c", po::value()->default_value(kDEFAULT_CONFIG_PATH), "configuration file") ("ng-web-server,w", "Use ng-web-server") ("migrate", po::value(), "start migration helper") + ("verify", "Checks the validity of config values") ; // clang-format on po::positional_options_description positional; @@ -75,6 +76,9 @@ CliArgs::parse(int argc, char const* argv[]) return Action{Action::Migrate{.configPath = std::move(configPath), .subCmd = MigrateSubCmd::migration(opt)}}; } + if (parsed.count("verify") != 0u) + return Action{Action::VerifyConfig{.configPath = std::move(configPath)}}; + return Action{Action::Run{.configPath = std::move(configPath), .useNgWebServer = parsed.count("ng-web-server") != 0} }; } diff --git a/src/app/CliArgs.hpp b/src/app/CliArgs.hpp index 2142ad3f6..8f2a8c359 100644 --- a/src/app/CliArgs.hpp +++ b/src/app/CliArgs.hpp @@ -59,6 +59,11 @@ public: MigrateSubCmd subCmd; }; + /** @brief Verify Config action. */ + struct VerifyConfig { + std::string configPath; + }; + /** * @brief Construct an action from a Run. * @@ -66,7 +71,7 @@ public: */ template requires std::is_same_v or std::is_same_v or - std::is_same_v + std::is_same_v or std::is_same_v explicit Action(ActionType&& action) : action_(std::forward(action)) { } @@ -86,7 +91,7 @@ public: } private: - std::variant action_; + std::variant action_; }; /** diff --git a/src/app/ClioApplication.cpp b/src/app/ClioApplication.cpp index 6a64206f4..bc5d0df1a 100644 --- a/src/app/ClioApplication.cpp +++ b/src/app/ClioApplication.cpp @@ -19,6 +19,7 @@ #include "app/ClioApplication.hpp" +#include "app/Stopper.hpp" #include "app/WebHandlers.hpp" #include "data/AmendmentCenter.hpp" #include "data/BackendFactory.hpp" @@ -26,6 +27,7 @@ #include "etl/LoadBalancer.hpp" #include "etl/NetworkValidatedLedgers.hpp" #include "feed/SubscriptionManager.hpp" +#include "migration/MigrationInspectorFactory.hpp" #include "rpc/Counters.hpp" #include "rpc/RPCEngine.hpp" #include "rpc/WorkQueue.hpp" @@ -83,6 +85,7 @@ ClioApplication::ClioApplication(util::config::ClioConfigDefinition const& confi { LOG(util::LogService::info()) << "Clio version: " << util::build::getClioFullVersionString(); PrometheusService::init(config); + signalsHandler_.subscribeToStop([this]() { appStopper_.stop(); }); } int @@ -103,6 +106,16 @@ ClioApplication::run(bool const useNgWebServer) // Interface to the database auto backend = data::makeBackend(config_); + { + auto const migrationInspector = migration::makeMigrationInspector(config_, backend); + // Check if any migration is blocking Clio server starting. + if (migrationInspector->isBlockingClio() and backend->hardFetchLedgerRangeNoThrow()) { + LOG(util::LogService::error()) + << "Existing Migration is blocking Clio, Please complete the database migration first."; + return EXIT_FAILURE; + } + } + // Manages clients subscribed to streams auto subscriptions = feed::SubscriptionManager::makeSubscriptionManager(config_, backend); @@ -158,6 +171,10 @@ ClioApplication::run(bool const useNgWebServer) return EXIT_FAILURE; } + appStopper_.setOnStop( + Stopper::makeOnStopCallback(httpServer.value(), *balancer, *etl, *subscriptions, *backend, ioc) + ); + // Blocks until stopped. // When stopped, shared_ptrs fall out of scope // Calls destructors on all resources, and destructs in order diff --git a/src/app/ClioApplication.hpp b/src/app/ClioApplication.hpp index 65b8f5f81..1eac02aae 100644 --- a/src/app/ClioApplication.hpp +++ b/src/app/ClioApplication.hpp @@ -19,6 +19,7 @@ #pragma once +#include "app/Stopper.hpp" #include "util/SignalsHandler.hpp" #include "util/newconfig/ConfigDefinition.hpp" @@ -30,6 +31,7 @@ namespace app { class ClioApplication { util::config::ClioConfigDefinition const& config_; util::SignalsHandler signalsHandler_; + Stopper appStopper_; public: /** diff --git a/src/app/Stopper.cpp b/src/app/Stopper.cpp new file mode 100644 index 000000000..90da5cde3 --- /dev/null +++ b/src/app/Stopper.cpp @@ -0,0 +1,52 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#include "app/Stopper.hpp" + +#include + +#include +#include +#include + +namespace app { + +Stopper::~Stopper() +{ + if (worker_.joinable()) + worker_.join(); +} + +void +Stopper::setOnStop(std::function cb) +{ + boost::asio::spawn(ctx_, std::move(cb)); +} + +void +Stopper::stop() +{ + // Do nothing if worker_ is already running + if (worker_.joinable()) + return; + + worker_ = std::thread{[this]() { ctx_.run(); }}; +} + +} // namespace app diff --git a/src/app/Stopper.hpp b/src/app/Stopper.hpp new file mode 100644 index 000000000..b4faa1377 --- /dev/null +++ b/src/app/Stopper.hpp @@ -0,0 +1,118 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2024, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "data/BackendInterface.hpp" +#include "etl/ETLService.hpp" +#include "etl/LoadBalancer.hpp" +#include "feed/SubscriptionManagerInterface.hpp" +#include "util/CoroutineGroup.hpp" +#include "util/log/Logger.hpp" +#include "web/ng/Server.hpp" + +#include +#include +#include + +#include +#include + +namespace app { + +/** + * @brief Application stopper class. On stop it will create a new thread to run all the shutdown tasks. + */ +class Stopper { + boost::asio::io_context ctx_; + std::thread worker_; + +public: + /** + * @brief Destroy the Stopper object + */ + ~Stopper(); + + /** + * @brief Set the callback to be called when the application is stopped. + * + * @param cb The callback to be called on application stop. + */ + void + setOnStop(std::function cb); + + /** + * @brief Stop the application and run the shutdown tasks. + */ + void + stop(); + + /** + * @brief Create a callback to be called on application stop. + * + * @param server The server to stop. + * @param balancer The load balancer to stop. + * @param etl The ETL service to stop. + * @param subscriptions The subscription manager to stop. + * @param backend The backend to stop. + * @param ioc The io_context to stop. + * @return The callback to be called on application stop. + */ + template < + web::ng::SomeServer ServerType, + etl::SomeLoadBalancer LoadBalancerType, + etl::SomeETLService ETLServiceType> + static std::function + makeOnStopCallback( + ServerType& server, + LoadBalancerType& balancer, + ETLServiceType& etl, + feed::SubscriptionManagerInterface& subscriptions, + data::BackendInterface& backend, + boost::asio::io_context& ioc + ) + { + return [&](boost::asio::yield_context yield) { + util::CoroutineGroup coroutineGroup{yield}; + coroutineGroup.spawn(yield, [&server](auto innerYield) { + server.stop(innerYield); + LOG(util::LogService::info()) << "Server stopped"; + }); + coroutineGroup.spawn(yield, [&balancer](auto innerYield) { + balancer.stop(innerYield); + LOG(util::LogService::info()) << "LoadBalancer stopped"; + }); + coroutineGroup.asyncWait(yield); + + etl.stop(); + LOG(util::LogService::info()) << "ETL stopped"; + + subscriptions.stop(); + LOG(util::LogService::info()) << "SubscriptionManager stopped"; + + backend.waitForWritesToFinish(); + LOG(util::LogService::info()) << "Backend writes finished"; + + ioc.stop(); + LOG(util::LogService::info()) << "io_context stopped"; + }; + } +}; + +} // namespace app diff --git a/src/app/VerifyConfig.hpp b/src/app/VerifyConfig.hpp new file mode 100644 index 000000000..9182072ec --- /dev/null +++ b/src/app/VerifyConfig.hpp @@ -0,0 +1,58 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "util/newconfig/ConfigDefinition.hpp" +#include "util/newconfig/ConfigFileJson.hpp" + +#include +#include +#include + +namespace app { + +/** + * @brief Verifies user's config values are correct + * + * @param configPath The path to config + * @return true if config values are all correct, false otherwise + */ +inline bool +parseConfig(std::string_view configPath) +{ + using namespace util::config; + + auto const json = ConfigFileJson::makeConfigFileJson(configPath); + if (!json.has_value()) { + std::cerr << "Error parsing json from config: " << configPath << "\n" << json.error().error << std::endl; + return false; + } + auto const errors = gClioConfig.parse(json.value()); + if (errors.has_value()) { + for (auto const& err : errors.value()) { + std::cerr << "Issues found in provided config '" << configPath << "':\n"; + std::cerr << err.error << std::endl; + } + return false; + } + return true; +} + +} // namespace app diff --git a/src/data/AmendmentCenter.hpp b/src/data/AmendmentCenter.hpp index 0413f009a..777ba340b 100644 --- a/src/data/AmendmentCenter.hpp +++ b/src/data/AmendmentCenter.hpp @@ -132,6 +132,8 @@ struct Amendments { REGISTER(fixAMMv1_2); REGISTER(AMMClawback); REGISTER(Credentials); + REGISTER(DynamicNFT); + REGISTER(PermissionedDomains); // Obsolete but supported by libxrpl REGISTER(CryptoConditionsSuite); diff --git a/src/data/BackendInterface.hpp b/src/data/BackendInterface.hpp index 2d7e3ef61..48771bae1 100644 --- a/src/data/BackendInterface.hpp +++ b/src/data/BackendInterface.hpp @@ -683,6 +683,12 @@ public: bool finishWrites(std::uint32_t ledgerSequence); + /** + * @brief Wait for all pending writes to finish. + */ + virtual void + waitForWritesToFinish() = 0; + /** * @brief Mark the migration status of a migrator as Migrated in the database * diff --git a/src/data/CassandraBackend.hpp b/src/data/CassandraBackend.hpp index ba8aacd7f..65309a5fd 100644 --- a/src/data/CassandraBackend.hpp +++ b/src/data/CassandraBackend.hpp @@ -188,11 +188,16 @@ public: return {txns, {}}; } + void + waitForWritesToFinish() override + { + executor_.sync(); + } + bool doFinishWrites() override { - // wait for other threads to finish their writes - executor_.sync(); + waitForWritesToFinish(); if (!range_) { executor_.writeSync(schema_->updateLedgerRange, ledgerSequence_, false, ledgerSequence_); @@ -619,7 +624,6 @@ public: return seq; } LOG(log_.debug()) << "Could not fetch ledger object sequence - no rows"; - } else { LOG(log_.error()) << "Could not fetch ledger object sequence: " << res.error(); } @@ -943,28 +947,35 @@ public: statements.reserve(data.size() * 3); for (NFTsData const& record : data) { - statements.push_back( - schema_->insertNFT.bind(record.tokenID, record.ledgerSequence, record.owner, record.isBurned) - ); + if (!record.onlyUriChanged) { + statements.push_back( + schema_->insertNFT.bind(record.tokenID, record.ledgerSequence, record.owner, record.isBurned) + ); - // If `uri` is set (and it can be set to an empty uri), we know this - // is a net-new NFT. That is, this NFT has not been seen before by - // us _OR_ it is in the extreme edge case of a re-minted NFT ID with - // the same NFT ID as an already-burned token. In this case, we need - // to record the URI and link to the issuer_nf_tokens table. - if (record.uri) { - statements.push_back(schema_->insertIssuerNFT.bind( - ripple::nft::getIssuer(record.tokenID), - static_cast(ripple::nft::getTaxon(record.tokenID)), - record.tokenID - )); + // If `uri` is set (and it can be set to an empty uri), we know this + // is a net-new NFT. That is, this NFT has not been seen before by + // us _OR_ it is in the extreme edge case of a re-minted NFT ID with + // the same NFT ID as an already-burned token. In this case, we need + // to record the URI and link to the issuer_nf_tokens table. + if (record.uri) { + statements.push_back(schema_->insertIssuerNFT.bind( + ripple::nft::getIssuer(record.tokenID), + static_cast(ripple::nft::getTaxon(record.tokenID)), + record.tokenID + )); + statements.push_back( + schema_->insertNFTURI.bind(record.tokenID, record.ledgerSequence, record.uri.value()) + ); + } + } else { + // only uri changed, we update the uri table only statements.push_back( schema_->insertNFTURI.bind(record.tokenID, record.ledgerSequence, record.uri.value()) ); } } - executor_.write(std::move(statements)); + executor_.writeEach(std::move(statements)); } void diff --git a/src/data/DBHelpers.hpp b/src/data/DBHelpers.hpp index 520228e38..cba1ec1ac 100644 --- a/src/data/DBHelpers.hpp +++ b/src/data/DBHelpers.hpp @@ -107,6 +107,7 @@ struct NFTsData { ripple::AccountID owner; std::optional uri; bool isBurned = false; + bool onlyUriChanged = false; // Whether only the URI was changed /** * @brief Construct a new NFTsData object @@ -170,6 +171,23 @@ struct NFTsData { : tokenID(tokenID), ledgerSequence(ledgerSequence), owner(owner), uri(uri) { } + + /** + * @brief Construct a new NFTsData object with only the URI changed + * + * @param tokenID The token ID + * @param meta The transaction metadata + * @param uri The new URI + * + */ + NFTsData(ripple::uint256 const& tokenID, ripple::TxMeta const& meta, ripple::Blob const& uri) + : tokenID(tokenID) + , ledgerSequence(meta.getLgrSeq()) + , transactionIndex(meta.getIndex()) + , uri(uri) + , onlyUriChanged(true) + { + } }; /** diff --git a/src/data/cassandra/Schema.hpp b/src/data/cassandra/Schema.hpp index 9816c53bd..4ec0356f9 100644 --- a/src/data/cassandra/Schema.hpp +++ b/src/data/cassandra/Schema.hpp @@ -74,7 +74,7 @@ public: 'class': 'SimpleStrategy', 'replication_factor': '{}' }} - AND durable_writes = true + AND durable_writes = True )", settingsProvider_.get().getKeyspace(), settingsProvider_.get().getReplicationFactor() @@ -472,7 +472,7 @@ public: R"( UPDATE {} SET sequence = ? - WHERE is_latest = false + WHERE is_latest = False )", qualifiedTableName(settingsProvider_.get(), "ledger_range") )); @@ -776,7 +776,7 @@ public: R"( SELECT sequence FROM {} - WHERE is_latest = true + WHERE is_latest = True )", qualifiedTableName(settingsProvider_.get(), "ledger_range") )); @@ -787,6 +787,7 @@ public: R"( SELECT sequence FROM {} + WHERE is_latest in (True, False) )", qualifiedTableName(settingsProvider_.get(), "ledger_range") )); diff --git a/src/data/cassandra/impl/AsyncExecutor.hpp b/src/data/cassandra/impl/AsyncExecutor.hpp index 25a306cb2..6ffb35032 100644 --- a/src/data/cassandra/impl/AsyncExecutor.hpp +++ b/src/data/cassandra/impl/AsyncExecutor.hpp @@ -23,6 +23,7 @@ #include "data/cassandra/Handle.hpp" #include "data/cassandra/Types.hpp" #include "data/cassandra/impl/RetryPolicy.hpp" +#include "util/Mutex.hpp" #include "util/log/Logger.hpp" #include @@ -64,8 +65,8 @@ class AsyncExecutor : public std::enable_shared_from_this future_; - std::mutex mtx_; + using OptionalFuture = std::optional; + util::Mutex future_; public: /** @@ -127,8 +128,8 @@ private: self = nullptr; // explicitly decrement refcount }; - std::scoped_lock const lck{mtx_}; - future_.emplace(handle.asyncExecute(data_, std::move(handler))); + auto future = future_.template lock(); + future->emplace(handle.asyncExecute(data_, std::move(handler))); } }; diff --git a/src/data/cassandra/impl/ExecutionStrategy.hpp b/src/data/cassandra/impl/ExecutionStrategy.hpp index 91ee8336e..0b95d3f8d 100644 --- a/src/data/cassandra/impl/ExecutionStrategy.hpp +++ b/src/data/cassandra/impl/ExecutionStrategy.hpp @@ -35,6 +35,7 @@ #include #include +#include #include #include #include @@ -192,10 +193,24 @@ public: template void write(PreparedStatementType const& preparedStatement, Args&&... args) + { + auto statement = preparedStatement.bind(std::forward(args)...); + write(std::move(statement)); + } + + /** + * @brief Non-blocking query execution used for writing data. + * + * Retries forever with retry policy specified by @ref AsyncExecutor + * + * @param statement Statement to execute + * @throw DatabaseTimeout on timeout + */ + void + write(StatementType&& statement) { auto const startTime = std::chrono::steady_clock::now(); - auto statement = preparedStatement.bind(std::forward(args)...); incrementOutstandingRequestCount(); counters_->registerWriteStarted(); @@ -251,6 +266,21 @@ public: }); } + /** + * @brief Non-blocking query execution used for writing data. Constrast with write, this method does not execute + * the statements in a batch. + * + * Retries forever with retry policy specified by @ref AsyncExecutor. + * + * @param statements Vector of statements to execute + * @throw DatabaseTimeout on timeout + */ + void + writeEach(std::vector&& statements) + { + std::ranges::for_each(std::move(statements), [this](auto& statement) { this->write(std::move(statement)); }); + } + /** * @brief Coroutine-based query execution used for reading data. * diff --git a/src/etl/CacheLoader.hpp b/src/etl/CacheLoader.hpp index 0d53688da..bb162aa73 100644 --- a/src/etl/CacheLoader.hpp +++ b/src/etl/CacheLoader.hpp @@ -130,7 +130,8 @@ public: void stop() noexcept { - loader_->stop(); + if (loader_ != nullptr) + loader_->stop(); } /** @@ -139,7 +140,8 @@ public: void wait() noexcept { - loader_->wait(); + if (loader_ != nullptr) + loader_->wait(); } }; diff --git a/src/etl/ETLService.cpp b/src/etl/ETLService.cpp index 3807efed8..2b35a96ac 100644 --- a/src/etl/ETLService.cpp +++ b/src/etl/ETLService.cpp @@ -134,7 +134,7 @@ ETLService::monitor() } } catch (std::runtime_error const& e) { LOG(log_.fatal()) << "Failed to load initial ledger: " << e.what(); - amendmentBlockHandler_.onAmendmentBlock(); + amendmentBlockHandler_.notifyAmendmentBlocked(); return; } diff --git a/src/etl/ETLService.hpp b/src/etl/ETLService.hpp index 1dbbd94c6..f0e2a2fc3 100644 --- a/src/etl/ETLService.hpp +++ b/src/etl/ETLService.hpp @@ -42,6 +42,7 @@ #include #include +#include #include #include #include @@ -58,6 +59,16 @@ struct NFTsData; */ namespace etl { +/** + * @brief A tag class to help identify ETLService in templated code. + */ +struct ETLServiceTag { + virtual ~ETLServiceTag() = default; +}; + +template +concept SomeETLService = std::derived_from; + /** * @brief This class is responsible for continuously extracting data from a p2p node, and writing that data to the * databases. @@ -71,7 +82,7 @@ namespace etl { * the others will fall back to monitoring/publishing. In this sense, this class dynamically transitions from monitoring * to writing and from writing to monitoring, based on the activity of other processes running on different machines. */ -class ETLService { +class ETLService : public ETLServiceTag { // TODO: make these template parameters in ETLService using LoadBalancerType = LoadBalancer; using DataPipeType = etl::impl::ExtractionDataPipe; @@ -159,10 +170,20 @@ public: /** * @brief Stops components and joins worker thread. */ - ~ETLService() + ~ETLService() override { - LOG(log_.info()) << "onStop called"; - LOG(log_.debug()) << "Stopping Reporting ETL"; + if (not state_.isStopping) + stop(); + } + + /** + * @brief Stop the ETL service. + * @note This method blocks until the ETL service has stopped. + */ + void + stop() + { + LOG(log_.info()) << "Stop called"; state_.isStopping = true; cacheLoader_.stop(); diff --git a/src/etl/LoadBalancer.cpp b/src/etl/LoadBalancer.cpp index 6e961933c..7dc0d1925 100644 --- a/src/etl/LoadBalancer.cpp +++ b/src/etl/LoadBalancer.cpp @@ -26,6 +26,7 @@ #include "feed/SubscriptionManagerInterface.hpp" #include "rpc/Errors.hpp" #include "util/Assert.hpp" +#include "util/CoroutineGroup.hpp" #include "util/Random.hpp" #include "util/ResponseExpirationCache.hpp" #include "util/log/Logger.hpp" @@ -336,6 +337,16 @@ LoadBalancer::getETLState() noexcept return etlState_; } +void +LoadBalancer::stop(boost::asio::yield_context yield) +{ + util::CoroutineGroup group{yield}; + std::ranges::for_each(sources_, [&group, yield](auto& source) { + group.spawn(yield, [&source](boost::asio::yield_context innerYield) { source->stop(innerYield); }); + }); + group.asyncWait(yield); +} + void LoadBalancer::chooseForwardingSource() { diff --git a/src/etl/LoadBalancer.hpp b/src/etl/LoadBalancer.hpp index 60f56a487..cfe01d019 100644 --- a/src/etl/LoadBalancer.hpp +++ b/src/etl/LoadBalancer.hpp @@ -41,6 +41,7 @@ #include #include +#include #include #include #include @@ -51,6 +52,16 @@ namespace etl { +/** + * @brief A tag class to help identify LoadBalancer in templated code. + */ +struct LoadBalancerTag { + virtual ~LoadBalancerTag() = default; +}; + +template +concept SomeLoadBalancer = std::derived_from; + /** * @brief This class is used to manage connections to transaction processing processes. * @@ -58,7 +69,7 @@ namespace etl { * which ledgers have been validated by the network, and the range of ledgers each etl source has). This class also * allows requests for ledger data to be load balanced across all possible ETL sources. */ -class LoadBalancer { +class LoadBalancer : public LoadBalancerTag { public: using RawLedgerObjectType = org::xrpl::rpc::v1::RawLedgerObject; using GetLedgerResponseType = org::xrpl::rpc::v1::GetLedgerResponse; @@ -132,7 +143,7 @@ public: SourceFactory sourceFactory = makeSource ); - ~LoadBalancer(); + ~LoadBalancer() override; /** * @brief Load the initial ledger, writing data to the queue. @@ -203,6 +214,15 @@ public: std::optional getETLState() noexcept; + /** + * @brief Stop the load balancer. This will stop all subscription sources. + * @note This function will asynchronously wait for all sources to stop. + * + * @param yield The coroutine context + */ + void + stop(boost::asio::yield_context yield); + private: /** * @brief Execute a function on a randomly selected source. diff --git a/src/etl/NFTHelpers.cpp b/src/etl/NFTHelpers.cpp index be3d739d8..f7e90e25f 100644 --- a/src/etl/NFTHelpers.cpp +++ b/src/etl/NFTHelpers.cpp @@ -47,6 +47,17 @@ namespace etl { +std::pair, std::optional> +getNftokenModifyData(ripple::TxMeta const& txMeta, ripple::STTx const& sttx) +{ + auto const tokenID = sttx.getFieldH256(ripple::sfNFTokenID); + // note: sfURI is optional, if it is absent, we will update the uri as empty string + return { + {NFTTransactionsData(sttx.getFieldH256(ripple::sfNFTokenID), txMeta, sttx.getTransactionID())}, + NFTsData(tokenID, txMeta, sttx.getFieldVL(ripple::sfURI)) + }; +} + std::pair, std::optional> getNFTokenMintData(ripple::TxMeta const& txMeta, ripple::STTx const& sttx) { @@ -166,7 +177,7 @@ getNFTokenBurnData(ripple::TxMeta const& txMeta, ripple::STTx const& sttx) node.peekAtField(ripple::sfPreviousFields).downcast(); if (previousFields.isFieldPresent(ripple::sfNFTokens)) prevNFTs = previousFields.getFieldArray(ripple::sfNFTokens); - } else if (!prevNFTs && node.getFName() == ripple::sfDeletedNode) { + } else if (node.getFName() == ripple::sfDeletedNode) { prevNFTs = node.peekAtField(ripple::sfFinalFields).downcast().getFieldArray(ripple::sfNFTokens); } @@ -336,6 +347,9 @@ getNFTDataFromTx(ripple::TxMeta const& txMeta, ripple::STTx const& sttx) case ripple::TxType::ttNFTOKEN_CREATE_OFFER: return getNFTokenCreateOfferData(txMeta, sttx); + case ripple::TxType::ttNFTOKEN_MODIFY: + return getNftokenModifyData(txMeta, sttx); + default: return {{}, {}}; } diff --git a/src/etl/NFTHelpers.hpp b/src/etl/NFTHelpers.hpp index c60eb9bf8..ef8cdb74b 100644 --- a/src/etl/NFTHelpers.hpp +++ b/src/etl/NFTHelpers.hpp @@ -33,6 +33,16 @@ namespace etl { +/** + * @brief Get the NFT URI change data from a NFToken Modify transaction + * + * @param txMeta Transaction metadata + * @param sttx The transaction + * @return NFT URI change data as a pair of transactions and optional NFTsData + */ +std::pair, std::optional> +getNftokenModifyData(ripple::TxMeta const& txMeta, ripple::STTx const& sttx); + /** * @brief Get the NFT Token mint data from a transaction * diff --git a/src/etl/Source.hpp b/src/etl/Source.hpp index e50849720..91d9bf817 100644 --- a/src/etl/Source.hpp +++ b/src/etl/Source.hpp @@ -65,6 +65,15 @@ public: virtual void run() = 0; + /** + * @brief Stop Source. + * @note This method will asynchronously wait for source to be stopped. + * + * @param yield The coroutine context. + */ + virtual void + stop(boost::asio::yield_context yield) = 0; + /** * @brief Check if source is connected * diff --git a/src/etl/impl/AmendmentBlockHandler.cpp b/src/etl/impl/AmendmentBlockHandler.cpp index 9b732bf6b..d4899f807 100644 --- a/src/etl/impl/AmendmentBlockHandler.cpp +++ b/src/etl/impl/AmendmentBlockHandler.cpp @@ -47,7 +47,7 @@ AmendmentBlockHandler::AmendmentBlockHandler( } void -AmendmentBlockHandler::onAmendmentBlock() +AmendmentBlockHandler::notifyAmendmentBlocked() { state_.get().isAmendmentBlocked = true; repeat_.start(interval_, action_); diff --git a/src/etl/impl/AmendmentBlockHandler.hpp b/src/etl/impl/AmendmentBlockHandler.hpp index b50d909b2..a49e75cb7 100644 --- a/src/etl/impl/AmendmentBlockHandler.hpp +++ b/src/etl/impl/AmendmentBlockHandler.hpp @@ -53,7 +53,7 @@ public: ); void - onAmendmentBlock(); + notifyAmendmentBlocked(); }; } // namespace etl::impl diff --git a/src/etl/impl/LedgerLoader.hpp b/src/etl/impl/LedgerLoader.hpp index 753b211bd..e6d88ee8e 100644 --- a/src/etl/impl/LedgerLoader.hpp +++ b/src/etl/impl/LedgerLoader.hpp @@ -57,6 +57,7 @@ struct FormattedTransactionsData { std::vector nfTokenTxData; std::vector nfTokensData; std::vector mptHoldersData; + std::vector nfTokenURIChanges; }; namespace etl::impl { @@ -111,6 +112,7 @@ public: { FormattedTransactionsData result; + std::vector nfTokenURIChanges; for (auto& txn : *(data.mutable_transactions_list()->mutable_transactions())) { std::string* raw = txn.mutable_transaction_blob(); @@ -123,8 +125,15 @@ public: auto const [nftTxs, maybeNFT] = getNFTDataFromTx(txMeta, sttx); result.nfTokenTxData.insert(result.nfTokenTxData.end(), nftTxs.begin(), nftTxs.end()); - if (maybeNFT) - result.nfTokensData.push_back(*maybeNFT); + + // We need to unique the URI changes separately, in case the URI changes are discarded + if (maybeNFT) { + if (maybeNFT->onlyUriChanged) { + nfTokenURIChanges.push_back(*maybeNFT); + } else { + result.nfTokensData.push_back(*maybeNFT); + } + } auto const maybeMPTHolder = getMPTHolderFromTx(txMeta, sttx); if (maybeMPTHolder) @@ -143,6 +152,10 @@ public: } result.nfTokensData = getUniqueNFTsDatas(result.nfTokensData); + nfTokenURIChanges = getUniqueNFTsDatas(nfTokenURIChanges); + + // Put uri change at the end to ensure the uri not overwritten + result.nfTokensData.insert(result.nfTokensData.end(), nfTokenURIChanges.begin(), nfTokenURIChanges.end()); return result; } diff --git a/src/etl/impl/SourceImpl.hpp b/src/etl/impl/SourceImpl.hpp index 7302bfa4c..3f2f25a33 100644 --- a/src/etl/impl/SourceImpl.hpp +++ b/src/etl/impl/SourceImpl.hpp @@ -102,6 +102,12 @@ public: subscriptionSource_->run(); } + void + stop(boost::asio::yield_context yield) final + { + subscriptionSource_->stop(yield); + } + /** * @brief Check if source is connected * diff --git a/src/etl/impl/SubscriptionSource.cpp b/src/etl/impl/SubscriptionSource.cpp index 1684cd23f..48d498b8c 100644 --- a/src/etl/impl/SubscriptionSource.cpp +++ b/src/etl/impl/SubscriptionSource.cpp @@ -35,7 +35,6 @@ #include #include #include -#include #include #include #include @@ -49,7 +48,6 @@ #include #include #include -#include #include #include #include @@ -92,15 +90,6 @@ SubscriptionSource::SubscriptionSource( .setConnectionTimeout(wsTimeout_); } -SubscriptionSource::~SubscriptionSource() -{ - stop(); - retry_.cancel(); - - if (runFuture_.valid()) - runFuture_.wait(); -} - void SubscriptionSource::run() { @@ -157,59 +146,53 @@ SubscriptionSource::validatedRange() const } void -SubscriptionSource::stop() +SubscriptionSource::stop(boost::asio::yield_context yield) { stop_ = true; + stopHelper_.asyncWaitForStop(yield); } void SubscriptionSource::subscribe() { - runFuture_ = boost::asio::spawn( - strand_, - [this, _ = boost::asio::make_work_guard(strand_)](boost::asio::yield_context yield) { - auto connection = wsConnectionBuilder_.connect(yield); - if (not connection) { - handleError(connection.error(), yield); - return; - } - + boost::asio::spawn(strand_, [this, _ = boost::asio::make_work_guard(strand_)](boost::asio::yield_context yield) { + if (auto connection = wsConnectionBuilder_.connect(yield); connection) { wsConnection_ = std::move(connection).value(); + } else { + handleError(connection.error(), yield); + return; + } - auto const& subscribeCommand = getSubscribeCommandJson(); - auto const writeErrorOpt = wsConnection_->write(subscribeCommand, yield, wsTimeout_); - if (writeErrorOpt) { - handleError(writeErrorOpt.value(), yield); + auto const& subscribeCommand = getSubscribeCommandJson(); + + if (auto const writeErrorOpt = wsConnection_->write(subscribeCommand, yield, wsTimeout_); writeErrorOpt) { + handleError(writeErrorOpt.value(), yield); + return; + } + + isConnected_ = true; + LOG(log_.info()) << "Connected"; + onConnect_(); + + retry_.reset(); + + while (!stop_) { + auto const message = wsConnection_->read(yield, wsTimeout_); + if (not message) { + handleError(message.error(), yield); return; } - isConnected_ = true; - LOG(log_.info()) << "Connected"; - onConnect_(); - - retry_.reset(); - - while (!stop_) { - auto const message = wsConnection_->read(yield, wsTimeout_); - if (not message) { - handleError(message.error(), yield); - return; - } - - auto const handleErrorOpt = handleMessage(message.value()); - if (handleErrorOpt) { - handleError(handleErrorOpt.value(), yield); - return; - } + if (auto const handleErrorOpt = handleMessage(message.value()); handleErrorOpt) { + handleError(handleErrorOpt.value(), yield); + return; } - // Close the connection - handleError( - util::requests::RequestError{"Subscription source stopped", boost::asio::error::operation_aborted}, - yield - ); - }, - boost::asio::use_future - ); + } + // Close the connection + handleError( + util::requests::RequestError{"Subscription source stopped", boost::asio::error::operation_aborted}, yield + ); + }); } std::optional @@ -299,6 +282,8 @@ SubscriptionSource::handleError(util::requests::RequestError const& error, boost logError(error); if (not stop_) { retry_.retry([this] { subscribe(); }); + } else { + stopHelper_.readyToStop(); } } diff --git a/src/etl/impl/SubscriptionSource.hpp b/src/etl/impl/SubscriptionSource.hpp index 77bdf4b0e..d57cb76b6 100644 --- a/src/etl/impl/SubscriptionSource.hpp +++ b/src/etl/impl/SubscriptionSource.hpp @@ -24,6 +24,7 @@ #include "feed/SubscriptionManagerInterface.hpp" #include "util/Mutex.hpp" #include "util/Retry.hpp" +#include "util/StopHelper.hpp" #include "util/log/Logger.hpp" #include "util/prometheus/Gauge.hpp" #include "util/requests/Types.hpp" @@ -39,7 +40,6 @@ #include #include #include -#include #include #include #include @@ -50,6 +50,7 @@ namespace etl::impl { /** * @brief This class is used to subscribe to a source of ledger data and forward it to the subscription manager. + * @note This class is safe to delete only if io_context is stopped. */ class SubscriptionSource { public: @@ -89,7 +90,7 @@ private: std::reference_wrapper lastMessageTimeSecondsSinceEpoch_; - std::future runFuture_; + util::StopHelper stopHelper_; static constexpr std::chrono::seconds kWS_TIMEOUT{30}; static constexpr std::chrono::seconds kRETRY_MAX_DELAY{30}; @@ -124,13 +125,6 @@ public: std::chrono::steady_clock::duration const retryDelay = SubscriptionSource::kRETRY_DELAY ); - /** - * @brief Destroy the Subscription Source object - * - * @note This will block to wait for all the async operations to complete. io_context must be still running - */ - ~SubscriptionSource(); - /** * @brief Run the source */ @@ -192,7 +186,7 @@ public: * @brief Stop the source. The source will complete already scheduled operations but will not schedule new ones */ void - stop(); + stop(boost::asio::yield_context yield); private: void diff --git a/src/etl/impl/Transformer.hpp b/src/etl/impl/Transformer.hpp index 834c3130d..b1d33b054 100644 --- a/src/etl/impl/Transformer.hpp +++ b/src/etl/impl/Transformer.hpp @@ -203,7 +203,7 @@ private: } catch (std::runtime_error const& e) { LOG(log_.fatal()) << "Failed to build next ledger: " << e.what(); - amendmentBlockHandler_.get().onAmendmentBlock(); + amendmentBlockHandler_.get().notifyAmendmentBlocked(); return {ripple::LedgerHeader{}, false}; } diff --git a/src/etlng/AmendmentBlockHandlerInterface.hpp b/src/etlng/AmendmentBlockHandlerInterface.hpp new file mode 100644 index 000000000..985d0e9cb --- /dev/null +++ b/src/etlng/AmendmentBlockHandlerInterface.hpp @@ -0,0 +1,37 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2024, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +namespace etlng { + +/** + * @brief The interface of a handler for amendment blocking + */ +struct AmendmentBlockHandlerInterface { + virtual ~AmendmentBlockHandlerInterface() = default; + + /** + * @brief The function to call once an amendment block has been discovered + */ + virtual void + notifyAmendmentBlocked() = 0; +}; + +} // namespace etlng diff --git a/src/etlng/CMakeLists.txt b/src/etlng/CMakeLists.txt index 5d9e29158..57fafcdfc 100644 --- a/src/etlng/CMakeLists.txt +++ b/src/etlng/CMakeLists.txt @@ -1,5 +1,8 @@ add_library(clio_etlng) -target_sources(clio_etlng PRIVATE impl/AsyncGrpcCall.cpp impl/Extraction.cpp impl/GrpcSource.cpp) +target_sources( + clio_etlng PRIVATE impl/AmendmentBlockHandler.cpp impl/AsyncGrpcCall.cpp impl/Extraction.cpp impl/GrpcSource.cpp + impl/Loading.cpp impl/TaskManager.cpp +) target_link_libraries(clio_etlng PUBLIC clio_data) diff --git a/src/etlng/LedgerPublisherInterface.hpp b/src/etlng/LedgerPublisherInterface.hpp new file mode 100644 index 000000000..e2d414214 --- /dev/null +++ b/src/etlng/LedgerPublisherInterface.hpp @@ -0,0 +1,49 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include +#include +#include + +namespace etlng { + +/** + * @brief The interface of a scheduler for the extraction proccess + */ +struct LedgerPublisherInterface { + virtual ~LedgerPublisherInterface() = default; + + /** + * @brief Publish the ledger by its sequence number + * + * @param seq The sequence number of the ledger + * @param maxAttempts The maximum number of attempts to publish the ledger; no limit if nullopt + * @param attemptsDelay The delay between attempts + */ + virtual void + publish( + uint32_t seq, + std::optional maxAttempts, + std::chrono::steady_clock::duration attemptsDelay = std::chrono::seconds{1} + ) = 0; +}; + +} // namespace etlng diff --git a/src/etlng/LoadBalancerInterface.hpp b/src/etlng/LoadBalancerInterface.hpp new file mode 100644 index 000000000..3466a6e79 --- /dev/null +++ b/src/etlng/LoadBalancerInterface.hpp @@ -0,0 +1,134 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2024, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +/** @file */ +#pragma once + +#include "etl/ETLState.hpp" +#include "etlng/InitialLoadObserverInterface.hpp" +#include "rpc/Errors.hpp" + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace etlng { + +/** + * @brief An interface for LoadBalancer + */ +class LoadBalancerInterface { +public: + using RawLedgerObjectType = org::xrpl::rpc::v1::RawLedgerObject; + using GetLedgerResponseType = org::xrpl::rpc::v1::GetLedgerResponse; + using OptionalGetLedgerResponseType = std::optional; + + virtual ~LoadBalancerInterface() = default; + + /** + * @brief Load the initial ledger, writing data to the queue. + * @note This function will retry indefinitely until the ledger is downloaded. + * + * @param sequence Sequence of ledger to download + * @param loader InitialLoadObserverInterface implementation + * @param retryAfter Time to wait between retries (2 seconds by default) + * @return A std::vector The ledger data + */ + virtual std::vector + loadInitialLedger( + uint32_t sequence, + etlng::InitialLoadObserverInterface& loader, + std::chrono::steady_clock::duration retryAfter = std::chrono::seconds{2} + ) = 0; + + /** + * @brief Load the initial ledger, writing data to the queue. + * @note This function will retry indefinitely until the ledger is downloaded. + * + * @param sequence Sequence of ledger to download + * @param retryAfter Time to wait between retries (2 seconds by default) + * @return A std::vector The ledger data + */ + virtual std::vector + loadInitialLedger(uint32_t sequence, std::chrono::steady_clock::duration retryAfter = std::chrono::seconds{2}) = 0; + + /** + * @brief Fetch data for a specific ledger. + * + * This function will continuously try to fetch data for the specified ledger until the fetch succeeds, the ledger + * is found in the database, or the server is shutting down. + * + * @param ledgerSequence Sequence of the ledger to fetch + * @param getObjects Whether to get the account state diff between this ledger and the prior one + * @param getObjectNeighbors Whether to request object neighbors + * @param retryAfter Time to wait between retries (2 seconds by default) + * @return The extracted data, if extraction was successful. If the ledger was found + * in the database or the server is shutting down, the optional will be empty + */ + virtual OptionalGetLedgerResponseType + fetchLedger( + uint32_t ledgerSequence, + bool getObjects, + bool getObjectNeighbors, + std::chrono::steady_clock::duration retryAfter = std::chrono::seconds{2} + ) = 0; + + /** + * @brief Represent the state of this load balancer as a JSON object + * + * @return JSON representation of the state of this load balancer. + */ + virtual boost::json::value + toJson() const = 0; + + /** + * @brief Forward a JSON RPC request to a randomly selected rippled node. + * + * @param request JSON-RPC request to forward + * @param clientIp The IP address of the peer, if known + * @param isAdmin Whether the request is from an admin + * @param yield The coroutine context + * @return Response received from rippled node as JSON object on success or error on failure + */ + virtual std::expected + forwardToRippled( + boost::json::object const& request, + std::optional const& clientIp, + bool isAdmin, + boost::asio::yield_context yield + ) = 0; + + /** + * @brief Return state of ETL nodes. + * @return ETL state, nullopt if etl nodes not available + */ + virtual std::optional + getETLState() noexcept = 0; +}; + +} // namespace etlng diff --git a/src/etlng/LoaderInterface.hpp b/src/etlng/LoaderInterface.hpp new file mode 100644 index 000000000..929d71679 --- /dev/null +++ b/src/etlng/LoaderInterface.hpp @@ -0,0 +1,52 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2024, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "etlng/Models.hpp" + +#include + +#include + +namespace etlng { + +/** + * @brief An interface for a ETL Loader + */ +struct LoaderInterface { + virtual ~LoaderInterface() = default; + + /** + * @brief Load ledger data + * @param data The data to load + */ + virtual void + load(model::LedgerData const& data) = 0; + + /** + * @brief Load the initial ledger + * @param data The data to load + * @return Optional ledger header + */ + virtual std::optional + loadInitialLedger(model::LedgerData const& data) = 0; +}; + +} // namespace etlng diff --git a/src/etlng/Models.hpp b/src/etlng/Models.hpp index 4e8aa6825..253ba8030 100644 --- a/src/etlng/Models.hpp +++ b/src/etlng/Models.hpp @@ -29,6 +29,7 @@ #include #include #include +#include #include #include @@ -79,6 +80,23 @@ struct Transaction { ripple::uint256 id; std::string key; // key is the above id as a string of 32 characters ripple::TxType type; + + /** + * @brief Compares Transaction objects to each other without considering sttx and meta fields + * @param other The Transaction to compare to + * @return true if transaction is equivalent; false otherwise + */ + bool + operator==(Transaction const& other) const + { + return raw == other.raw // + and metaRaw == other.metaRaw // + and sttx.getTransactionID() == other.sttx.getTransactionID() // + and meta.getTxID() == other.meta.getTxID() // + and id == other.id // + and key == other.key // + and type == other.type; + } }; /** @@ -103,6 +121,9 @@ struct Object { std::string predecessor; ModType type; + + bool + operator==(Object const&) const = default; }; /** @@ -111,6 +132,9 @@ struct Object { struct BookSuccessor { std::string firstBook; std::string bookBase; + + bool + operator==(BookSuccessor const&) const = default; }; /** @@ -125,6 +149,45 @@ struct LedgerData { ripple::LedgerHeader header; std::string rawHeader; uint32_t seq; + + /** + * @brief Compares LedgerData objects to each other without considering the header field + * @param other The LedgerData to compare to + * @return true if data is equivalent; false otherwise + */ + bool + operator==(LedgerData const& other) const + { + auto const serialized = [](auto const& hdr) { + ripple::Serializer ser; + ripple::addRaw(hdr, ser); + return ser.getString(); + }; + + return transactions == other.transactions // + and objects == other.objects // + and successors == other.successors // + and edgeKeys == other.edgeKeys // + and serialized(header) == serialized(other.header) // + and rawHeader == other.rawHeader // + and seq == other.seq; + } +}; + +/** + * @brief Represents a task for the extractors + */ +struct Task { + /** + * @brief Represents the priority of the task + */ + enum class Priority : uint8_t { + Lower = 0u, + Higher = 1u, + }; + + Priority priority; + uint32_t seq; }; } // namespace etlng::model diff --git a/src/etlng/SchedulerInterface.hpp b/src/etlng/SchedulerInterface.hpp new file mode 100644 index 000000000..606757b60 --- /dev/null +++ b/src/etlng/SchedulerInterface.hpp @@ -0,0 +1,42 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "etlng/Models.hpp" + +#include + +namespace etlng { + +/** + * @brief The interface of a scheduler for the extraction proccess + */ +struct SchedulerInterface { + virtual ~SchedulerInterface() = default; + + /** + * @brief Attempt to obtain the next task + * @return A task if one exists; std::nullopt otherwise + */ + [[nodiscard]] virtual std::optional + next() = 0; +}; + +} // namespace etlng diff --git a/src/etlng/impl/AmendmentBlockHandler.cpp b/src/etlng/impl/AmendmentBlockHandler.cpp new file mode 100644 index 000000000..502e31812 --- /dev/null +++ b/src/etlng/impl/AmendmentBlockHandler.cpp @@ -0,0 +1,56 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2024, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#include "etlng/impl/AmendmentBlockHandler.hpp" + +#include "etl/SystemState.hpp" +#include "util/async/AnyExecutionContext.hpp" +#include "util/log/Logger.hpp" + +#include +#include +#include + +namespace etlng::impl { + +AmendmentBlockHandler::ActionType const AmendmentBlockHandler::kDEFAULT_AMENDMENT_BLOCK_ACTION = []() { + static util::Logger const log{"ETL"}; // NOLINT(readability-identifier-naming) + LOG(log.fatal()) << "Can't process new ledgers: The current ETL source is not compatible with the version of " + << "the libxrpl Clio is currently using. Please upgrade Clio to a newer version."; +}; + +AmendmentBlockHandler::AmendmentBlockHandler( + util::async::AnyExecutionContext&& ctx, + etl::SystemState& state, + std::chrono::steady_clock::duration interval, + ActionType action +) + : state_{std::ref(state)}, interval_{interval}, ctx_{std::move(ctx)}, action_{std::move(action)} +{ +} + +void +AmendmentBlockHandler::notifyAmendmentBlocked() +{ + state_.get().isAmendmentBlocked = true; + if (not operation_.has_value()) + operation_.emplace(ctx_.executeRepeatedly(interval_, action_)); +} + +} // namespace etlng::impl diff --git a/src/etlng/impl/AmendmentBlockHandler.hpp b/src/etlng/impl/AmendmentBlockHandler.hpp new file mode 100644 index 000000000..0bece8e79 --- /dev/null +++ b/src/etlng/impl/AmendmentBlockHandler.hpp @@ -0,0 +1,69 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2024, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "etl/SystemState.hpp" +#include "etlng/AmendmentBlockHandlerInterface.hpp" +#include "util/async/AnyExecutionContext.hpp" +#include "util/async/AnyOperation.hpp" + +#include +#include +#include + +#include +#include +#include + +namespace etlng::impl { + +class AmendmentBlockHandler : public AmendmentBlockHandlerInterface { +public: + using ActionType = std::function; + +private: + std::reference_wrapper state_; + std::chrono::steady_clock::duration interval_; + util::async::AnyExecutionContext ctx_; + std::optional> operation_; + + ActionType action_; + +public: + static ActionType const kDEFAULT_AMENDMENT_BLOCK_ACTION; + + AmendmentBlockHandler( + util::async::AnyExecutionContext&& ctx, + etl::SystemState& state, + std::chrono::steady_clock::duration interval = std::chrono::seconds{1}, + ActionType action = kDEFAULT_AMENDMENT_BLOCK_ACTION + ); + + ~AmendmentBlockHandler() override + { + if (operation_.has_value()) + operation_.value().abort(); + } + + void + notifyAmendmentBlocked() override; +}; + +} // namespace etlng::impl diff --git a/src/etlng/impl/AsyncGrpcCall.cpp b/src/etlng/impl/AsyncGrpcCall.cpp index 6fd44dced..87e132503 100644 --- a/src/etlng/impl/AsyncGrpcCall.cpp +++ b/src/etlng/impl/AsyncGrpcCall.cpp @@ -87,14 +87,14 @@ AsyncGrpcCall::process( if (abort) { LOG(log_.error()) << "AsyncGrpcCall aborted"; - return CallStatus::ERRORED; + return CallStatus::Errored; } if (!status_.ok()) { LOG(log_.error()) << "AsyncGrpcCall status_ not ok: code = " << status_.error_code() << " message = " << status_.error_message(); - return CallStatus::ERRORED; + return CallStatus::Errored; } if (!next_->is_unlimited()) { @@ -141,7 +141,7 @@ AsyncGrpcCall::process( predecessorKey_ = lastKey_; // but for ongoing onInitialObjects calls we need to pass along the key we left // off at so that we can link the two lists correctly - return more ? CallStatus::MORE : CallStatus::DONE; + return more ? CallStatus::More : CallStatus::Done; } void diff --git a/src/etlng/impl/AsyncGrpcCall.hpp b/src/etlng/impl/AsyncGrpcCall.hpp index 8cb3edb84..044323257 100644 --- a/src/etlng/impl/AsyncGrpcCall.hpp +++ b/src/etlng/impl/AsyncGrpcCall.hpp @@ -38,7 +38,7 @@ namespace etlng::impl { class AsyncGrpcCall { public: - enum class CallStatus { MORE, DONE, ERRORED }; + enum class CallStatus { More, Done, Errored }; using RequestType = org::xrpl::rpc::v1::GetLedgerDataRequest; using ResponseType = org::xrpl::rpc::v1::GetLedgerDataResponse; using StubType = org::xrpl::rpc::v1::XRPLedgerAPIService::Stub; diff --git a/src/etlng/impl/GrpcSource.cpp b/src/etlng/impl/GrpcSource.cpp index adcbba7e7..a537bcd08 100644 --- a/src/etlng/impl/GrpcSource.cpp +++ b/src/etlng/impl/GrpcSource.cpp @@ -139,7 +139,7 @@ GrpcSource::loadInitialLedger( LOG(log_.trace()) << "Marker prefix = " << ptr->getMarkerPrefix(); auto result = ptr->process(stub_, queue, observer, abort); - if (result != AsyncGrpcCall::CallStatus::MORE) { + if (result != AsyncGrpcCall::CallStatus::More) { ++numFinished; LOG(log_.debug()) << "Finished a marker. Current number of finished = " << numFinished; @@ -147,7 +147,7 @@ GrpcSource::loadInitialLedger( edgeKeys.push_back(std::move(lastKey)); } - if (result == AsyncGrpcCall::CallStatus::ERRORED) + if (result == AsyncGrpcCall::CallStatus::Errored) abort = true; } diff --git a/src/etlng/impl/Loading.cpp b/src/etlng/impl/Loading.cpp new file mode 100644 index 000000000..e8608dbed --- /dev/null +++ b/src/etlng/impl/Loading.cpp @@ -0,0 +1,111 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#include "etlng/impl/Loading.hpp" + +#include "data/BackendInterface.hpp" +#include "etl/LedgerFetcherInterface.hpp" +#include "etl/impl/LedgerLoader.hpp" +#include "etlng/AmendmentBlockHandlerInterface.hpp" +#include "etlng/Models.hpp" +#include "etlng/RegistryInterface.hpp" +#include "util/Assert.hpp" +#include "util/LedgerUtils.hpp" +#include "util/Profiler.hpp" +#include "util/log/Logger.hpp" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace etlng::impl { + +Loader::Loader( + std::shared_ptr backend, + std::shared_ptr fetcher, + std::shared_ptr registry, + std::shared_ptr amendmentBlockHandler +) + : backend_(std::move(backend)) + , fetcher_(std::move(fetcher)) + , registry_(std::move(registry)) + , amendmentBlockHandler_(std::move(amendmentBlockHandler)) +{ +} + +void +Loader::load(model::LedgerData const& data) +{ + try { + // perform cache updates and all writes from extensions + registry_->dispatch(data); + + auto [success, duration] = + ::util::timed>([&]() { return backend_->finishWrites(data.seq); }); + LOG(log_.info()) << "Finished writes to DB for " << data.seq << ": " << (success ? "YES" : "NO") << "; took " + << duration; + } catch (std::runtime_error const& e) { + LOG(log_.fatal()) << "Failed to load " << data.seq << ": " << e.what(); + amendmentBlockHandler_->notifyAmendmentBlocked(); + } +}; + +void +Loader::onInitialLoadGotMoreObjects( + uint32_t seq, + std::vector const& data, + std::optional lastKey +) +{ + LOG(log_.debug()) << "On initial load: got more objects for seq " << seq << ". size = " << data.size(); + registry_->dispatchInitialObjects( + seq, data, std::move(lastKey).value_or(std::string{}) // TODO: perhaps use optional all the way to extensions? + ); +} + +std::optional +Loader::loadInitialLedger(model::LedgerData const& data) +{ + // check that database is actually empty + auto rng = backend_->hardFetchLedgerRangeNoThrow(); + if (rng) { + ASSERT(false, "Database is not empty"); + return std::nullopt; + } + + LOG(log_.debug()) << "Deserialized ledger header. " << ::util::toString(data.header); + + auto seconds = ::util::timed([this, &data]() { registry_->dispatchInitialData(data); }); + LOG(log_.info()) << "Dispatching initial data and submitting all writes took " << seconds << " seconds."; + + backend_->finishWrites(data.seq); + LOG(log_.debug()) << "Loaded initial ledger"; + + return {data.header}; +} + +} // namespace etlng::impl diff --git a/src/etlng/impl/Loading.hpp b/src/etlng/impl/Loading.hpp new file mode 100644 index 000000000..435dcb4e4 --- /dev/null +++ b/src/etlng/impl/Loading.hpp @@ -0,0 +1,85 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2024, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "data/BackendInterface.hpp" +#include "etl/LedgerFetcherInterface.hpp" +#include "etl/impl/LedgerLoader.hpp" +#include "etlng/AmendmentBlockHandlerInterface.hpp" +#include "etlng/InitialLoadObserverInterface.hpp" +#include "etlng/LoaderInterface.hpp" +#include "etlng/Models.hpp" +#include "etlng/RegistryInterface.hpp" +#include "util/log/Logger.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace etlng::impl { + +class Loader : public LoaderInterface, public InitialLoadObserverInterface { + std::shared_ptr backend_; + std::shared_ptr fetcher_; + std::shared_ptr registry_; + std::shared_ptr amendmentBlockHandler_; + + util::Logger log_{"ETL"}; + +public: + using RawLedgerObjectType = org::xrpl::rpc::v1::RawLedgerObject; + using GetLedgerResponseType = org::xrpl::rpc::v1::GetLedgerResponse; + using OptionalGetLedgerResponseType = std::optional; + + Loader( + std::shared_ptr backend, + std::shared_ptr fetcher, + std::shared_ptr registry, + std::shared_ptr amendmentBlockHandler + ); + + void + load(model::LedgerData const& data) override; + + void + onInitialLoadGotMoreObjects( + uint32_t seq, + std::vector const& data, + std::optional lastKey + ) override; + + std::optional + loadInitialLedger(model::LedgerData const& data) override; +}; + +} // namespace etlng::impl diff --git a/src/etlng/impl/Scheduling.hpp b/src/etlng/impl/Scheduling.hpp new file mode 100644 index 000000000..7e16aa9c0 --- /dev/null +++ b/src/etlng/impl/Scheduling.hpp @@ -0,0 +1,151 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "etl/NetworkValidatedLedgersInterface.hpp" +#include "etlng/Models.hpp" +#include "etlng/SchedulerInterface.hpp" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace etlng::impl { + +template +concept SomeScheduler = std::is_base_of_v>; + +class ForwardScheduler : public SchedulerInterface { + std::reference_wrapper ledgers_; + + uint32_t startSeq_; + std::optional maxSeq_; + std::atomic_uint32_t seq_; + +public: + ForwardScheduler(ForwardScheduler const& other) + : ledgers_(other.ledgers_), startSeq_(other.startSeq_), maxSeq_(other.maxSeq_), seq_(other.seq_.load()) + { + } + + ForwardScheduler( + std::reference_wrapper ledgers, + uint32_t startSeq, + std::optional maxSeq = std::nullopt + ) + : ledgers_(ledgers), startSeq_(startSeq), maxSeq_(maxSeq), seq_(startSeq) + { + } + + [[nodiscard]] std::optional + next() override + { + static constexpr auto kMAX = std::numeric_limits::max(); + uint32_t currentSeq = seq_; + + if (ledgers_.get().getMostRecent() >= currentSeq) { + while (currentSeq < maxSeq_.value_or(kMAX)) { + if (seq_.compare_exchange_weak(currentSeq, currentSeq + 1u, std::memory_order_acq_rel)) { + return {{.priority = model::Task::Priority::Higher, .seq = currentSeq}}; + } + } + } + + return std::nullopt; + } +}; + +class BackfillScheduler : public SchedulerInterface { + uint32_t startSeq_; + uint32_t minSeq_ = 0u; + + std::atomic_uint32_t seq_; + +public: + BackfillScheduler(BackfillScheduler const& other) + : startSeq_(other.startSeq_), minSeq_(other.minSeq_), seq_(other.seq_.load()) + { + } + + BackfillScheduler(uint32_t startSeq, std::optional minSeq = std::nullopt) + : startSeq_(startSeq), minSeq_(minSeq.value_or(0)), seq_(startSeq) + { + } + + [[nodiscard]] std::optional + next() override + { + uint32_t currentSeq = seq_; + while (currentSeq > minSeq_) { + if (seq_.compare_exchange_weak(currentSeq, currentSeq - 1u, std::memory_order_acq_rel)) { + return {{.priority = model::Task::Priority::Lower, .seq = currentSeq}}; + } + } + + return std::nullopt; + } +}; + +template +class SchedulerChain : public SchedulerInterface { + std::tuple schedulers_; + +public: + template + requires(std::is_same_v and ...) + SchedulerChain(Ts&&... schedulers) : schedulers_(std::forward(schedulers)...) + { + } + + [[nodiscard]] std::optional + next() override + { + std::optional task; + auto const expand = [&](auto& s) { + if (task.has_value()) + return false; + + task = s.next(); + return task.has_value(); + }; + + std::apply([&expand](auto&&... xs) { (... || expand(xs)); }, schedulers_); + + return task; + } +}; + +static auto +makeScheduler(SomeScheduler auto&&... schedulers) +{ + return std::make_unique...>>( + std::forward(schedulers)... + ); +} + +} // namespace etlng::impl diff --git a/src/etlng/impl/TaskManager.cpp b/src/etlng/impl/TaskManager.cpp new file mode 100644 index 000000000..a5fc20ce0 --- /dev/null +++ b/src/etlng/impl/TaskManager.cpp @@ -0,0 +1,141 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#include "etlng/impl/TaskManager.hpp" + +#include "etlng/ExtractorInterface.hpp" +#include "etlng/LoaderInterface.hpp" +#include "etlng/Models.hpp" +#include "etlng/SchedulerInterface.hpp" +#include "util/async/AnyExecutionContext.hpp" +#include "util/async/AnyOperation.hpp" +#include "util/async/AnyStrand.hpp" +#include "util/log/Logger.hpp" + +#include +#include +#include +#include +#include +#include +#include + +namespace etlng::impl { + +TaskManager::TaskManager( + util::async::AnyExecutionContext&& ctx, + std::reference_wrapper scheduler, + std::reference_wrapper extractor, + std::reference_wrapper loader +) + : ctx_(std::move(ctx)), schedulers_(scheduler), extractor_(extractor), loader_(loader) +{ +} + +TaskManager::~TaskManager() +{ + stop(); +} + +void +TaskManager::run(Settings settings) +{ + static constexpr auto kQUEUE_SIZE_LIMIT = 2048uz; + + auto schedulingStrand = ctx_.makeStrand(); + PriorityQueue queue(ctx_.makeStrand(), kQUEUE_SIZE_LIMIT); + + LOG(log_.debug()) << "Starting task manager...\n"; + + extractors_.reserve(settings.numExtractors); + for ([[maybe_unused]] auto _ : std::views::iota(0uz, settings.numExtractors)) + extractors_.push_back(spawnExtractor(schedulingStrand, queue)); + + loaders_.reserve(settings.numLoaders); + for ([[maybe_unused]] auto _ : std::views::iota(0uz, settings.numLoaders)) + loaders_.push_back(spawnLoader(queue)); + + wait(); + LOG(log_.debug()) << "All finished in task manager..\n"; +} + +util::async::AnyOperation +TaskManager::spawnExtractor(util::async::AnyStrand& strand, PriorityQueue& queue) +{ + // TODO: these values may be extracted to config later and/or need to be fine-tuned on a realistic system + static constexpr auto kDELAY_BETWEEN_ATTEMPTS = std::chrono::milliseconds{100u}; + static constexpr auto kDELAY_BETWEEN_ENQUEUE_ATTEMPTS = std::chrono::milliseconds{1u}; + + return strand.execute([this, &queue](auto stopRequested) { + while (not stopRequested) { + if (auto task = schedulers_.get().next(); task.has_value()) { + if (auto maybeBatch = extractor_.get().extractLedgerWithDiff(task->seq); maybeBatch.has_value()) { + LOG(log_.debug()) << "Adding data after extracting diff"; + while (not queue.enqueue(*maybeBatch)) { + // TODO (https://github.com/XRPLF/clio/issues/1852) + std::this_thread::sleep_for(kDELAY_BETWEEN_ENQUEUE_ATTEMPTS); + + if (stopRequested) + break; + } + } else { + // TODO: how do we signal to the loaders that it's time to shutdown? some special task? + break; // TODO: handle server shutdown or other node took over ETL + } + } else { + // TODO (https://github.com/XRPLF/clio/issues/1852) + std::this_thread::sleep_for(kDELAY_BETWEEN_ATTEMPTS); + } + } + }); +} + +util::async::AnyOperation +TaskManager::spawnLoader(PriorityQueue& queue) +{ + return ctx_.execute([this, &queue](auto stopRequested) { + while (not stopRequested) { + // TODO (https://github.com/XRPLF/clio/issues/66): does not tell the loader whether it's out of order or not + if (auto data = queue.dequeue(); data.has_value()) + loader_.get().load(*data); + } + }); +} + +void +TaskManager::wait() +{ + for (auto& extractor : extractors_) + extractor.wait(); + for (auto& loader : loaders_) + loader.wait(); +} + +void +TaskManager::stop() +{ + for (auto& extractor : extractors_) + extractor.abort(); + for (auto& loader : loaders_) + loader.abort(); + + wait(); +} + +} // namespace etlng::impl diff --git a/src/etlng/impl/TaskManager.hpp b/src/etlng/impl/TaskManager.hpp new file mode 100644 index 000000000..0047513bc --- /dev/null +++ b/src/etlng/impl/TaskManager.hpp @@ -0,0 +1,94 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "etlng/ExtractorInterface.hpp" +#include "etlng/LoaderInterface.hpp" +#include "etlng/Models.hpp" +#include "etlng/SchedulerInterface.hpp" +#include "util/StrandedPriorityQueue.hpp" +#include "util/async/AnyExecutionContext.hpp" +#include "util/async/AnyOperation.hpp" +#include "util/async/AnyStrand.hpp" +#include "util/log/Logger.hpp" + +#include + +#include +#include +#include + +namespace etlng::impl { + +class TaskManager { + util::async::AnyExecutionContext ctx_; + std::reference_wrapper schedulers_; + std::reference_wrapper extractor_; + std::reference_wrapper loader_; + + std::vector> extractors_; + std::vector> loaders_; + + util::Logger log_{"ETL"}; + + struct ReverseOrderComparator { + [[nodiscard]] bool + operator()(model::LedgerData const& lhs, model::LedgerData const& rhs) const noexcept + { + return lhs.seq > rhs.seq; + } + }; + +public: + struct Settings { + size_t numExtractors; /**< number of extraction tasks */ + size_t numLoaders; /**< number of loading tasks */ + }; + + // reverse order loading is needed (i.e. start with oldest seq in forward fill buffer) + using PriorityQueue = util::StrandedPriorityQueue; + + TaskManager( + util::async::AnyExecutionContext&& ctx, + std::reference_wrapper scheduler, + std::reference_wrapper extractor, + std::reference_wrapper loader + ); + + ~TaskManager(); + + void + run(Settings settings); + + void + stop(); + +private: + void + wait(); + + [[nodiscard]] util::async::AnyOperation + spawnExtractor(util::async::AnyStrand& strand, PriorityQueue& queue); + + [[nodiscard]] util::async::AnyOperation + spawnLoader(PriorityQueue& queue); +}; + +} // namespace etlng::impl diff --git a/src/feed/SubscriptionManager.cpp b/src/feed/SubscriptionManager.cpp index 4f4e69c85..8c169be24 100644 --- a/src/feed/SubscriptionManager.cpp +++ b/src/feed/SubscriptionManager.cpp @@ -50,7 +50,7 @@ void SubscriptionManager::pubBookChanges( ripple::LedgerHeader const& lgrInfo, std::vector const& transactions -) const +) { bookChangesFeed_.pub(lgrInfo, transactions); } @@ -111,7 +111,7 @@ SubscriptionManager::pubLedger( ripple::Fees const& fees, std::string const& ledgerRange, std::uint32_t const txnCount -) const +) { ledgerFeed_.pub(lgrInfo, fees, ledgerRange, txnCount); } @@ -129,7 +129,7 @@ SubscriptionManager::unsubManifest(SubscriberSharedPtr const& subscriber) } void -SubscriptionManager::forwardManifest(boost::json::object const& manifestJson) const +SubscriptionManager::forwardManifest(boost::json::object const& manifestJson) { manifestFeed_.pub(manifestJson); } @@ -147,7 +147,7 @@ SubscriptionManager::unsubValidation(SubscriberSharedPtr const& subscriber) } void -SubscriptionManager::forwardValidation(boost::json::object const& validationJson) const +SubscriptionManager::forwardValidation(boost::json::object const& validationJson) { validationsFeed_.pub(validationJson); } diff --git a/src/feed/SubscriptionManager.hpp b/src/feed/SubscriptionManager.hpp index 096b7fb2d..c51ea5270 100644 --- a/src/feed/SubscriptionManager.hpp +++ b/src/feed/SubscriptionManager.hpp @@ -115,6 +115,15 @@ public: * @brief Destructor of the SubscriptionManager object. It will block until all running jobs finished. */ ~SubscriptionManager() override + { + stop(); + } + + /** + * @brief Stop the SubscriptionManager and wait for all jobs to finish. + */ + void + stop() override { ctx_.stop(); ctx_.join(); @@ -141,7 +150,7 @@ public: */ void pubBookChanges(ripple::LedgerHeader const& lgrInfo, std::vector const& transactions) - const final; + final; /** * @brief Subscribe to the proposed transactions feed. @@ -209,7 +218,7 @@ public: ripple::Fees const& fees, std::string const& ledgerRange, std::uint32_t txnCount - ) const final; + ) final; /** * @brief Subscribe to the manifest feed. @@ -230,7 +239,7 @@ public: * @param manifestJson The manifest json to forward. */ void - forwardManifest(boost::json::object const& manifestJson) const final; + forwardManifest(boost::json::object const& manifestJson) final; /** * @brief Subscribe to the validation feed. @@ -251,7 +260,7 @@ public: * @param validationJson The validation feed json to forward. */ void - forwardValidation(boost::json::object const& validationJson) const final; + forwardValidation(boost::json::object const& validationJson) final; /** * @brief Subscribe to the transactions feed. diff --git a/src/feed/SubscriptionManagerInterface.hpp b/src/feed/SubscriptionManagerInterface.hpp index c627bc53c..4339e5951 100644 --- a/src/feed/SubscriptionManagerInterface.hpp +++ b/src/feed/SubscriptionManagerInterface.hpp @@ -45,6 +45,12 @@ class SubscriptionManagerInterface { public: virtual ~SubscriptionManagerInterface() = default; + /** + * @brief Stop the SubscriptionManager and wait for all jobs to finish. + */ + virtual void + stop() = 0; + /** * @brief Subscribe to the book changes feed. * @param subscriber @@ -65,8 +71,10 @@ public: * @param transactions The transactions in the current ledger. */ virtual void - pubBookChanges(ripple::LedgerHeader const& lgrInfo, std::vector const& transactions) - const = 0; + pubBookChanges( + ripple::LedgerHeader const& lgrInfo, + std::vector const& transactions + ) = 0; /** * @brief Subscribe to the proposed transactions feed. @@ -135,7 +143,7 @@ public: ripple::Fees const& fees, std::string const& ledgerRange, std::uint32_t txnCount - ) const = 0; + ) = 0; /** * @brief Subscribe to the manifest feed. @@ -156,7 +164,7 @@ public: * @param manifestJson The manifest json to forward. */ virtual void - forwardManifest(boost::json::object const& manifestJson) const = 0; + forwardManifest(boost::json::object const& manifestJson) = 0; /** * @brief Subscribe to the validation feed. @@ -177,7 +185,7 @@ public: * @param validationJson The validation feed json to forward. */ virtual void - forwardValidation(boost::json::object const& validationJson) const = 0; + forwardValidation(boost::json::object const& validationJson) = 0; /** * @brief Subscribe to the transactions feed. diff --git a/src/feed/impl/BookChangesFeed.hpp b/src/feed/impl/BookChangesFeed.hpp index b8a05dad0..4d2693959 100644 --- a/src/feed/impl/BookChangesFeed.hpp +++ b/src/feed/impl/BookChangesFeed.hpp @@ -48,7 +48,7 @@ struct BookChangesFeed : public SingleFeedBase { * @param transactions The transactions that were included in the ledger. */ void - pub(ripple::LedgerHeader const& lgrInfo, std::vector const& transactions) const + pub(ripple::LedgerHeader const& lgrInfo, std::vector const& transactions) { SingleFeedBase::pub(boost::json::serialize(rpc::computeBookChanges(lgrInfo, transactions))); } diff --git a/src/feed/impl/ForwardFeed.hpp b/src/feed/impl/ForwardFeed.hpp index 39d04441b..80ad55151 100644 --- a/src/feed/impl/ForwardFeed.hpp +++ b/src/feed/impl/ForwardFeed.hpp @@ -36,7 +36,7 @@ struct ForwardFeed : public SingleFeedBase { * @brief Publishes the json object. */ void - pub(boost::json::object const& json) const + pub(boost::json::object const& json) { SingleFeedBase::pub(boost::json::serialize(json)); } diff --git a/src/feed/impl/LedgerFeed.cpp b/src/feed/impl/LedgerFeed.cpp index ee9220891..c0ef8e713 100644 --- a/src/feed/impl/LedgerFeed.cpp +++ b/src/feed/impl/LedgerFeed.cpp @@ -94,7 +94,7 @@ LedgerFeed::pub( ripple::Fees const& fees, std::string const& ledgerRange, std::uint32_t const txnCount -) const +) { SingleFeedBase::pub(boost::json::serialize(makeLedgerPubMessage(lgrInfo, fees, ledgerRange, txnCount))); } diff --git a/src/feed/impl/LedgerFeed.hpp b/src/feed/impl/LedgerFeed.hpp index 1550b97f8..2a1e419c9 100644 --- a/src/feed/impl/LedgerFeed.hpp +++ b/src/feed/impl/LedgerFeed.hpp @@ -76,7 +76,7 @@ public: pub(ripple::LedgerHeader const& lgrInfo, ripple::Fees const& fees, std::string const& ledgerRange, - std::uint32_t txnCount) const; + std::uint32_t txnCount); private: static boost::json::object diff --git a/src/feed/impl/SingleFeedBase.cpp b/src/feed/impl/SingleFeedBase.cpp index 4650cb267..ded998914 100644 --- a/src/feed/impl/SingleFeedBase.cpp +++ b/src/feed/impl/SingleFeedBase.cpp @@ -62,7 +62,7 @@ SingleFeedBase::unsub(SubscriberSharedPtr const& subscriber) } void -SingleFeedBase::pub(std::string msg) const +SingleFeedBase::pub(std::string msg) { [[maybe_unused]] auto task = strand_.execute([this, msg = std::move(msg)]() { auto const msgPtr = std::make_shared(msg); diff --git a/src/feed/impl/SingleFeedBase.hpp b/src/feed/impl/SingleFeedBase.hpp index 8fae3e976..d31eeb0c1 100644 --- a/src/feed/impl/SingleFeedBase.hpp +++ b/src/feed/impl/SingleFeedBase.hpp @@ -73,7 +73,7 @@ public: * @param msg The message. */ void - pub(std::string msg) const; + pub(std::string msg); /** * @brief Get the count of subscribers. diff --git a/src/feed/impl/TrackableSignal.hpp b/src/feed/impl/TrackableSignal.hpp index 0384d7dec..f190be3c1 100644 --- a/src/feed/impl/TrackableSignal.hpp +++ b/src/feed/impl/TrackableSignal.hpp @@ -19,6 +19,8 @@ #pragma once +#include "util/Mutex.hpp" + #include #include #include @@ -45,8 +47,8 @@ class TrackableSignal { // map of connection and signal connection, key is the pointer of the connection object // allow disconnect to be called in the destructor of the connection - std::unordered_map connections_; - mutable std::mutex mutex_; + using ConnectionsMap = std::unordered_map; + util::Mutex connections_; using SignalType = boost::signals2::signal; SignalType signal_; @@ -64,8 +66,8 @@ public: bool connectTrackableSlot(ConnectionSharedPtr const& trackable, std::function slot) { - std::scoped_lock const lk(mutex_); - if (connections_.contains(trackable.get())) { + auto connections = connections_.template lock(); + if (connections->contains(trackable.get())) { return false; } @@ -73,7 +75,7 @@ public: // the trackable's destructor. However, the trackable can not be destroied when the slot is being called // either. track_foreign will hold a weak_ptr to the connection, which makes sure the connection is valid when // the slot is called. - connections_.emplace( + connections->emplace( trackable.get(), signal_.connect(typename SignalType::slot_type(slot).track_foreign(trackable)) ); return true; @@ -89,10 +91,9 @@ public: bool disconnect(ConnectionPtr trackablePtr) { - std::scoped_lock const lk(mutex_); - if (connections_.contains(trackablePtr)) { - connections_[trackablePtr].disconnect(); - connections_.erase(trackablePtr); + if (auto connections = connections_.template lock(); connections->contains(trackablePtr)) { + connections->operator[](trackablePtr).disconnect(); + connections->erase(trackablePtr); return true; } return false; @@ -115,8 +116,7 @@ public: std::size_t count() const { - std::scoped_lock const lk(mutex_); - return connections_.size(); + return connections_.template lock()->size(); } }; } // namespace feed::impl diff --git a/src/feed/impl/TrackableSignalMap.hpp b/src/feed/impl/TrackableSignalMap.hpp index 38dd91be5..f9e3a06d6 100644 --- a/src/feed/impl/TrackableSignalMap.hpp +++ b/src/feed/impl/TrackableSignalMap.hpp @@ -20,6 +20,7 @@ #pragma once #include "feed/impl/TrackableSignal.hpp" +#include "util/Mutex.hpp" #include @@ -49,8 +50,8 @@ class TrackableSignalMap { using ConnectionPtr = Session*; using ConnectionSharedPtr = std::shared_ptr; - mutable std::mutex mutex_; - std::unordered_map> signalsMap_; + using SignalsMap = std::unordered_map>; + util::Mutex signalsMap_; public: /** @@ -66,8 +67,8 @@ public: bool connectTrackableSlot(ConnectionSharedPtr const& trackable, Key const& key, std::function slot) { - std::scoped_lock const lk(mutex_); - return signalsMap_[key].connectTrackableSlot(trackable, slot); + auto map = signalsMap_.template lock(); + return map->operator[](key).connectTrackableSlot(trackable, slot); } /** @@ -80,14 +81,14 @@ public: bool disconnect(ConnectionPtr trackablePtr, Key const& key) { - std::scoped_lock const lk(mutex_); - if (!signalsMap_.contains(key)) + auto map = signalsMap_.template lock(); + if (!map->contains(key)) return false; - auto const disconnected = signalsMap_[key].disconnect(trackablePtr); + auto const disconnected = map->operator[](key).disconnect(trackablePtr); // clean the map if there is no connection left. - if (disconnected && signalsMap_[key].count() == 0) - signalsMap_.erase(key); + if (disconnected && map->operator[](key).count() == 0) + map->erase(key); return disconnected; } @@ -101,9 +102,9 @@ public: void emit(Key const& key, Args const&... args) { - std::scoped_lock const lk(mutex_); - if (signalsMap_.contains(key)) - signalsMap_[key].emit(args...); + auto map = signalsMap_.template lock(); + if (map->contains(key)) + map->operator[](key).emit(args...); } }; } // namespace feed::impl diff --git a/src/main/Main.cpp b/src/main/Main.cpp index 6a817f0dc..4dd39e712 100644 --- a/src/main/Main.cpp +++ b/src/main/Main.cpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above @@ -19,12 +19,12 @@ #include "app/CliArgs.hpp" #include "app/ClioApplication.hpp" +#include "app/VerifyConfig.hpp" #include "migration/MigrationApplication.hpp" #include "rpc/common/impl/HandlerProvider.hpp" #include "util/TerminationHandler.hpp" #include "util/log/Logger.hpp" #include "util/newconfig/ConfigDefinition.hpp" -#include "util/newconfig/ConfigFileJson.hpp" #include #include @@ -40,34 +40,26 @@ try { auto const action = app::CliArgs::parse(argc, argv); return action.apply( [](app::CliArgs::Action::Exit const& exit) { return exit.exitCode; }, + [](app::CliArgs::Action::VerifyConfig const& verify) { + if (app::parseConfig(verify.configPath)) { + std::cout << "Config " << verify.configPath << " is correct" + << "\n"; + return EXIT_SUCCESS; + } + return EXIT_FAILURE; + }, [](app::CliArgs::Action::Run const& run) { - auto const json = ConfigFileJson::makeConfigFileJson(run.configPath); - if (!json.has_value()) { - std::cerr << json.error().error << std::endl; + if (not app::parseConfig(run.configPath)) return EXIT_FAILURE; - } - auto const errors = gClioConfig.parse(json.value()); - if (errors.has_value()) { - for (auto const& err : errors.value()) - std::cerr << err.error << std::endl; - return EXIT_FAILURE; - } + util::LogService::init(gClioConfig); app::ClioApplication clio{gClioConfig}; return clio.run(run.useNgWebServer); }, [](app::CliArgs::Action::Migrate const& migrate) { - auto const json = ConfigFileJson::makeConfigFileJson(migrate.configPath); - if (!json.has_value()) { - std::cerr << json.error().error << std::endl; + if (not app::parseConfig(migrate.configPath)) return EXIT_FAILURE; - } - auto const errors = gClioConfig.parse(json.value()); - if (errors.has_value()) { - for (auto const& err : errors.value()) - std::cerr << err.error << std::endl; - return EXIT_FAILURE; - } + util::LogService::init(gClioConfig); app::MigratorApplication migrator{gClioConfig, migrate.subCmd}; return migrator.run(); diff --git a/src/migration/CMakeLists.txt b/src/migration/CMakeLists.txt index 95a18031b..bc4499aa7 100644 --- a/src/migration/CMakeLists.txt +++ b/src/migration/CMakeLists.txt @@ -5,4 +5,4 @@ target_sources( cassandra/impl/ObjectsAdapter.cpp cassandra/impl/TransactionsAdapter.cpp ) -target_link_libraries(clio_migration PRIVATE clio_util clio_etl) +target_link_libraries(clio_migration PRIVATE clio_util clio_data) diff --git a/src/migration/MigrationApplication.cpp b/src/migration/MigrationApplication.cpp index 5b05a848c..c628316d9 100644 --- a/src/migration/MigrationApplication.cpp +++ b/src/migration/MigrationApplication.cpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/MigrationApplication.hpp b/src/migration/MigrationApplication.hpp index 81d9cc077..a38687dca 100644 --- a/src/migration/MigrationApplication.hpp +++ b/src/migration/MigrationApplication.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/MigrationInspectorFactory.hpp b/src/migration/MigrationInspectorFactory.hpp new file mode 100644 index 000000000..2baa29e8b --- /dev/null +++ b/src/migration/MigrationInspectorFactory.hpp @@ -0,0 +1,65 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "data/BackendInterface.hpp" +#include "migration/MigrationInspectorInterface.hpp" +#include "migration/MigratiorStatus.hpp" +#include "migration/cassandra/CassandraMigrationManager.hpp" +#include "util/Assert.hpp" +#include "util/log/Logger.hpp" +#include "util/newconfig/ConfigDefinition.hpp" + +#include +#include + +#include +#include + +namespace migration { + +/** + * @brief A factory function that creates migration inspector instance and initializes the migration table if needed. + * + * @param config The config. + * @param backend The backend instance. It should be initialized before calling this function. + * @return A shared_ptr instance + */ +inline std::shared_ptr +makeMigrationInspector( + util::config::ClioConfigDefinition const& config, + std::shared_ptr const& backend +) +{ + ASSERT(backend != nullptr, "Backend is not initialized"); + + auto inspector = std::make_shared(backend); + + // Database is empty, we need to initialize the migration table if it is a writeable backend + if (not config.get("read_only") and not backend->hardFetchLedgerRangeNoThrow()) { + migration::MigratorStatus const migrated(migration::MigratorStatus::Migrated); + for (auto const& name : inspector->allMigratorsNames()) { + backend->writeMigratorStatus(name, migrated.toString()); + } + } + return inspector; +} + +} // namespace migration diff --git a/src/migration/MigrationInspectorInterface.hpp b/src/migration/MigrationInspectorInterface.hpp new file mode 100644 index 000000000..a3382bc0f --- /dev/null +++ b/src/migration/MigrationInspectorInterface.hpp @@ -0,0 +1,79 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "migration/MigratiorStatus.hpp" + +#include +#include +#include + +namespace migration { + +/** + * @brief The interface for the migration inspector.The Clio server application will use this interface to inspect + * the migration status. + */ +struct MigrationInspectorInterface { + virtual ~MigrationInspectorInterface() = default; + + /** + * @brief Get the status of all the migrators + * @return A vector of tuple, the first element is the migrator's name, the second element is the status of the + */ + virtual std::vector> + allMigratorsStatusPairs() const = 0; + + /** + * @brief Get all registered migrators' names + * + * @return A vector of migrators' names + */ + virtual std::vector + allMigratorsNames() const = 0; + + /** + * @brief Get the status of a migrator by its name + * + * @param name The migrator's name + * @return The status of the migrator + */ + virtual MigratorStatus + getMigratorStatusByName(std::string const& name) const = 0; + + /** + * @brief Get the description of a migrator by its name + * + * @param name The migrator's name + * @return The description of the migrator + */ + virtual std::string + getMigratorDescriptionByName(std::string const& name) const = 0; + + /** + * @brief Return if Clio server is blocked + * + * @return True if Clio server is blocked by migration, false otherwise + */ + virtual bool + isBlockingClio() const = 0; +}; + +} // namespace migration diff --git a/src/migration/MigrationManagerInterface.hpp b/src/migration/MigrationManagerInterface.hpp index 66eafef92..e76bfeb54 100644 --- a/src/migration/MigrationManagerInterface.hpp +++ b/src/migration/MigrationManagerInterface.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above @@ -19,59 +19,23 @@ #pragma once -#include "migration/MigratiorStatus.hpp" +#include "migration/MigrationInspectorInterface.hpp" #include -#include -#include namespace migration { /** - * @brief The interface for the migration manager. This interface is tend to be implemented for specific database. The - * application layer will use this interface to run the migrations. + * @brief The interface for the migration manager. The migration application layer will use this interface to run the + * migrations. Unlike the MigrationInspectorInterface which only provides the status of migration, this interface + * contains the acutal migration running method. */ -struct MigrationManagerInterface { - virtual ~MigrationManagerInterface() = default; - +struct MigrationManagerInterface : virtual public MigrationInspectorInterface { /** * @brief Run the the migration according to the given migrator's name */ virtual void runMigration(std::string const&) = 0; - - /** - * @brief Get the status of all the migrators - * @return A vector of tuple, the first element is the migrator's name, the second element is the status of the - */ - virtual std::vector> - allMigratorsStatusPairs() const = 0; - - /** - * @brief Get all registered migrators' names - * - * @return A vector of migrators' names - */ - virtual std::vector - allMigratorsNames() const = 0; - - /** - * @brief Get the status of a migrator by its name - * - * @param name The migrator's name - * @return The status of the migrator - */ - virtual MigratorStatus - getMigratorStatusByName(std::string const& name) const = 0; - - /** - * @brief Get the description of a migrator by its name - * - * @param name The migrator's name - * @return The description of the migrator - */ - virtual std::string - getMigratorDescriptionByName(std::string const& name) const = 0; }; } // namespace migration diff --git a/src/migration/MigratiorStatus.hpp b/src/migration/MigratiorStatus.hpp index c01b0fabc..747f7f0a8 100644 --- a/src/migration/MigratiorStatus.hpp +++ b/src/migration/MigratiorStatus.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/MigratorStatus.cpp b/src/migration/MigratorStatus.cpp index 75c66df95..cfc630377 100644 --- a/src/migration/MigratorStatus.cpp +++ b/src/migration/MigratorStatus.cpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/README.md b/src/migration/README.md index 087c08024..9c297a894 100644 --- a/src/migration/README.md +++ b/src/migration/README.md @@ -40,9 +40,11 @@ A migrator satisfies the `MigratorSpec`(impl/Spec.hpp) concept. It contains: -- A `name` which will be used to identify the migrator. User will refer this migrator in command-line tool by this name. The name needs to be different with other migrators, otherwise a compilation error will be raised. +- A `kNAME` which will be used to identify the migrator. User will refer this migrator in command-line tool by this name. The name needs to be different with other migrators, otherwise a compilation error will be raised. -- A `description` which is the detail information of the migrator. +- A `kDESCRIPTION` which is the detail information of the migrator. + +- An optional `kCAN_BLOCK_CLIO` which indicates whether the migrator can block the Clio server. If it's absent, the migrator can't block server. If there is a blocking migrator not completed, the Clio server will fail to start. - A static function `runMigration`, it will be called when user run `--migrate name`. It accepts two parameters: backend, which provides the DB operations interface, and cfg, which provides migration-related configuration. Each migrator can have its own configuration under `.migration` session. @@ -65,8 +67,8 @@ Most indexes are based on either ledger states or transactions. We provide the ` If you need to do full scan against other table, you can follow below steps: - Describe the table which needs full scan in a struct. It has to satisfy the `TableSpec`(cassandra/Spec.hpp) concept, containing static member: - Tuple type `Row`, it's the type of each field in a row. The order of types should match what database will return in a row. Key types should come first, followed by other field types sorted in alphabetical order. - - `PARTITION_KEY`, it's the name of the partition key of the table. - - `TABLE_NAME` + - `kPARTITION_KEY`, it's the name of the partition key of the table. + - `kTABLE_NAME` - Inherent from `FullTableScannerAdapterBase`. - Implement `onRowRead`, its parameter is the `Row` we defined. It's the callback function when a row is read. diff --git a/src/migration/cassandra/CassandraMigrationBackend.hpp b/src/migration/cassandra/CassandraMigrationBackend.hpp index f967d2f35..bbc39255a 100644 --- a/src/migration/cassandra/CassandraMigrationBackend.hpp +++ b/src/migration/cassandra/CassandraMigrationBackend.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/CassandraMigrationManager.hpp b/src/migration/cassandra/CassandraMigrationManager.hpp index e3a17a238..a658a8201 100644 --- a/src/migration/cassandra/CassandraMigrationManager.hpp +++ b/src/migration/cassandra/CassandraMigrationManager.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above @@ -19,20 +19,32 @@ #pragma once +#include "data/BackendInterface.hpp" #include "migration/cassandra/CassandraMigrationBackend.hpp" +#include "migration/impl/MigrationInspectorBase.hpp" #include "migration/impl/MigrationManagerBase.hpp" #include "migration/impl/MigratorsRegister.hpp" -namespace migration::cassandra { +namespace { // Register migrators here // MigratorsRegister template using CassandraSupportedMigrators = migration::impl::MigratorsRegister; -// Register with MigrationBackend which proceeds the migration -using MigrationProcesser = CassandraSupportedMigrators; +// Instantiates with the backend which supports actual migration running +using MigrationProcesser = CassandraSupportedMigrators; + +// Instantiates with backend interface, it doesn't support actual migration. But it can be used to inspect the migrators +// status +using MigrationQuerier = CassandraSupportedMigrators; + +} // namespace + +namespace migration::cassandra { + +using CassandraMigrationInspector = migration::impl::MigrationInspectorBase; -// The Cassandra migration manager using CassandraMigrationManager = migration::impl::MigrationManagerBase; + } // namespace migration::cassandra diff --git a/src/migration/cassandra/impl/CassandraMigrationSchema.hpp b/src/migration/cassandra/impl/CassandraMigrationSchema.hpp index d998a6638..56fde4985 100644 --- a/src/migration/cassandra/impl/CassandraMigrationSchema.hpp +++ b/src/migration/cassandra/impl/CassandraMigrationSchema.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/impl/FullTableScanner.hpp b/src/migration/cassandra/impl/FullTableScanner.hpp index 34261893f..8407760c0 100644 --- a/src/migration/cassandra/impl/FullTableScanner.hpp +++ b/src/migration/cassandra/impl/FullTableScanner.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/impl/FullTableScannerAdapterBase.hpp b/src/migration/cassandra/impl/FullTableScannerAdapterBase.hpp index c0dc291e7..5373af47e 100644 --- a/src/migration/cassandra/impl/FullTableScannerAdapterBase.hpp +++ b/src/migration/cassandra/impl/FullTableScannerAdapterBase.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/impl/ObjectsAdapter.cpp b/src/migration/cassandra/impl/ObjectsAdapter.cpp index 0a74a31d7..d7d60abbb 100644 --- a/src/migration/cassandra/impl/ObjectsAdapter.cpp +++ b/src/migration/cassandra/impl/ObjectsAdapter.cpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/impl/ObjectsAdapter.hpp b/src/migration/cassandra/impl/ObjectsAdapter.hpp index 91463e3be..6d4362588 100644 --- a/src/migration/cassandra/impl/ObjectsAdapter.hpp +++ b/src/migration/cassandra/impl/ObjectsAdapter.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/impl/Spec.hpp b/src/migration/cassandra/impl/Spec.hpp index c71956630..a7ed3748b 100644 --- a/src/migration/cassandra/impl/Spec.hpp +++ b/src/migration/cassandra/impl/Spec.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/impl/TransactionsAdapter.cpp b/src/migration/cassandra/impl/TransactionsAdapter.cpp index c7b90bbfa..9aa9915b7 100644 --- a/src/migration/cassandra/impl/TransactionsAdapter.cpp +++ b/src/migration/cassandra/impl/TransactionsAdapter.cpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/impl/TransactionsAdapter.hpp b/src/migration/cassandra/impl/TransactionsAdapter.hpp index 6bb27ee8a..ed1b92b2d 100644 --- a/src/migration/cassandra/impl/TransactionsAdapter.hpp +++ b/src/migration/cassandra/impl/TransactionsAdapter.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/cassandra/impl/Types.hpp b/src/migration/cassandra/impl/Types.hpp index 2d65956c4..df1755a0a 100644 --- a/src/migration/cassandra/impl/Types.hpp +++ b/src/migration/cassandra/impl/Types.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/impl/MigrationInspectorBase.hpp b/src/migration/impl/MigrationInspectorBase.hpp new file mode 100644 index 000000000..fe712ec6b --- /dev/null +++ b/src/migration/impl/MigrationInspectorBase.hpp @@ -0,0 +1,121 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include "migration/MigrationInspectorInterface.hpp" +#include "migration/MigratiorStatus.hpp" + +#include +#include +#include +#include +#include + +namespace migration::impl { + +/** + * @brief The migration inspector implementation for Cassandra. It will report the migration status for Cassandra + * database. + * + * @tparam SupportedMigrators The migrators resgister that contains all the migrators + */ +template +class MigrationInspectorBase : virtual public MigrationInspectorInterface { +protected: + SupportedMigrators migrators_; + +public: + /** + * @brief Construct a new Cassandra Migration Inspector object + * + * @param backend The backend of the Cassandra database + */ + explicit MigrationInspectorBase(std::shared_ptr backend) + : migrators_{std::move(backend)} + { + } + + /** + * @brief Get the status of all the migrators + * + * @return A vector of tuple, the first element is the migrator's name, the second element is the status of the + * migrator + */ + std::vector> + allMigratorsStatusPairs() const override + { + return migrators_.getMigratorsStatus(); + } + + /** + * @brief Get the status of a migrator by its name + * + * @param name The name of the migrator + * @return The status of the migrator + */ + MigratorStatus + getMigratorStatusByName(std::string const& name) const override + { + return migrators_.getMigratorStatus(name); + } + + /** + * @brief Get all registered migrators' names + * + * @return A vector of string, the names of all the migrators + */ + std::vector + allMigratorsNames() const override + { + auto const names = migrators_.getMigratorNames(); + return std::vector{names.begin(), names.end()}; + } + + /** + * @brief Get the description of a migrator by its name + * + * @param name The name of the migrator + * @return The description of the migrator + */ + std::string + getMigratorDescriptionByName(std::string const& name) const override + { + return migrators_.getMigratorDescription(name); + } + + /** + * @brief Return if there is uncomplete migrator blocking the server + * + * @return True if server is blocked, false otherwise + */ + bool + isBlockingClio() const override + { + return std::ranges::any_of(migrators_.getMigratorNames(), [&](auto const& migrator) { + if (auto canBlock = migrators_.canMigratorBlockClio(migrator); canBlock.has_value() and *canBlock and + migrators_.getMigratorStatus(std::string(migrator)) == MigratorStatus::Status::NotMigrated) { + return true; + } + return false; + }); + } +}; + +} // namespace migration::impl diff --git a/src/migration/impl/MigrationManagerBase.hpp b/src/migration/impl/MigrationManagerBase.hpp index c04c035ef..76c21231a 100644 --- a/src/migration/impl/MigrationManagerBase.hpp +++ b/src/migration/impl/MigrationManagerBase.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above @@ -20,14 +20,12 @@ #pragma once #include "migration/MigrationManagerInterface.hpp" -#include "migration/MigratiorStatus.hpp" +#include "migration/impl/MigrationInspectorBase.hpp" #include "util/newconfig/ObjectView.hpp" #include #include -#include #include -#include namespace migration::impl { @@ -38,8 +36,7 @@ namespace migration::impl { * @tparam SupportedMigrators The migrators resgister that contains all the migrators */ template -class MigrationManagerBase : public MigrationManagerInterface { - SupportedMigrators migrators_; +class MigrationManagerBase : public MigrationManagerInterface, public MigrationInspectorBase { // contains only migration related settings util::config::ObjectView config_; @@ -54,7 +51,7 @@ public: std::shared_ptr backend, util::config::ObjectView config ) - : migrators_{backend}, config_{std::move(config)} + : MigrationInspectorBase{std::move(backend)}, config_{std::move(config)} { } @@ -66,55 +63,7 @@ public: void runMigration(std::string const& name) override { - migrators_.runMigrator(name, config_); - } - - /** - * @brief Get the status of all the migrators - * - * @return A vector of tuple, the first element is the migrator's name, the second element is the status of the - * migrator - */ - std::vector> - allMigratorsStatusPairs() const override - { - return migrators_.getMigratorsStatus(); - } - - /** - * @brief Get the status of a migrator by its name - * - * @param name The name of the migrator - * @return The status of the migrator - */ - MigratorStatus - getMigratorStatusByName(std::string const& name) const override - { - return migrators_.getMigratorStatus(name); - } - - /** - * @brief Get all registered migrators' names - * - * @return A vector of string, the names of all the migrators - */ - std::vector - allMigratorsNames() const override - { - auto const names = migrators_.getMigratorNames(); - return std::vector{names.begin(), names.end()}; - } - - /** - * @brief Get the description of a migrator by its name - * - * @param name The name of the migrator - * @return The description of the migrator - */ - std::string - getMigratorDescriptionByName(std::string const& name) const override - { - return migrators_.getMigratorDescription(name); + this->migrators_.runMigrator(name, config_); } }; diff --git a/src/migration/impl/MigrationManagerFactory.cpp b/src/migration/impl/MigrationManagerFactory.cpp index af4c550f9..9ab5d2343 100644 --- a/src/migration/impl/MigrationManagerFactory.cpp +++ b/src/migration/impl/MigrationManagerFactory.cpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/impl/MigrationManagerFactory.hpp b/src/migration/impl/MigrationManagerFactory.hpp index 936ac59b6..0cca4997d 100644 --- a/src/migration/impl/MigrationManagerFactory.hpp +++ b/src/migration/impl/MigrationManagerFactory.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/migration/impl/MigratorsRegister.hpp b/src/migration/impl/MigratorsRegister.hpp index e01c6e0e6..95bbc783e 100644 --- a/src/migration/impl/MigratorsRegister.hpp +++ b/src/migration/impl/MigratorsRegister.hpp @@ -1,7 +1,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above @@ -22,6 +22,7 @@ #include "data/BackendInterface.hpp" #include "migration/MigratiorStatus.hpp" #include "migration/impl/Spec.hpp" +#include "util/Assert.hpp" #include "util/Concepts.hpp" #include "util/log/Logger.hpp" #include "util/newconfig/ObjectView.hpp" @@ -30,10 +31,12 @@ #include #include #include +#include #include #include #include #include +#include #include namespace migration::impl { @@ -47,6 +50,11 @@ concept MigrationBackend = requires { requires std::same_as concept BackendMatchAllMigrators = (MigrationBackend && ...); +template +concept HasCanBlockClio = requires(T t) { + { t.kCAN_BLOCK_CLIO }; +}; + /** *@brief The register of migrators. It will dispatch the migration to the corresponding migrator. It also *hold the shared pointer of backend, which is used by the migrators. @@ -81,6 +89,23 @@ class MigratorsRegister { return (T::kNAME == targetName) ? T::kDESCRIPTION : ""; } + template + static constexpr bool + canBlockClioHelper(std::string_view targetName) + { + if (targetName == First::kNAME) { + if constexpr (HasCanBlockClio) { + return First::kCAN_BLOCK_CLIO; + } + return false; + } + if constexpr (sizeof...(Rest) > 0) { + return canBlockClioHelper(targetName); + } + ASSERT(false, "The migrator name is not found"); + std::unreachable(); + } + public: /** * @brief The backend type which is used by the migrators @@ -179,6 +204,27 @@ public: return result.empty() ? "No Description" : result; } } + + /** + * @brief Return if the given migrator can block Clio server + * + * @param name The migrator's name + * @return std::nullopt if the migrator name is not found, or a boolean value indicating whether the migrator is + * blocking Clio server. + */ + std::optional + canMigratorBlockClio(std::string_view name) const + { + if constexpr (sizeof...(MigratorType) == 0) { + return std::nullopt; + } else { + auto const migratiors = getMigratorNames(); + if (std::ranges::find(migratiors, name) == migratiors.end()) + return std::nullopt; + + return canBlockClioHelper(name); + } + } }; } // namespace migration::impl diff --git a/src/migration/impl/Spec.hpp b/src/migration/impl/Spec.hpp index 5d6c15f72..d9d379441 100644 --- a/src/migration/impl/Spec.hpp +++ b/src/migration/impl/Spec.hpp @@ -2,7 +2,7 @@ //------------------------------------------------------------------------------ /* This file is part of clio: https://github.com/XRPLF/clio - Copyright (c) 2022-2024, the clio developers. + Copyright (c) 2024, the clio developers. Permission to use, copy, modify, and distribute this software for any purpose with or without fee is hereby granted, provided that the above diff --git a/src/rpc/Errors.cpp b/src/rpc/Errors.cpp index 16e91ae83..2609fc881 100644 --- a/src/rpc/Errors.cpp +++ b/src/rpc/Errors.cpp @@ -79,7 +79,6 @@ getErrorInfo(ClioError code) {.code = ClioError::RpcMalformedRequest, .error = "malformedRequest", .message = "Malformed request."}, {.code = ClioError::RpcMalformedOwner, .error = "malformedOwner", .message = "Malformed owner."}, {.code = ClioError::RpcMalformedAddress, .error = "malformedAddress", .message = "Malformed address."}, - {.code = ClioError::RpcInvalidHotWallet, .error = "invalidHotWallet", .message = "Invalid hot wallet."}, {.code = ClioError::RpcUnknownOption, .error = "unknownOption", .message = "Unknown option."}, {.code = ClioError::RpcFieldNotFoundTransaction, .error = "fieldNotFoundTransaction", @@ -90,6 +89,7 @@ getErrorInfo(ClioError code) {.code = ClioError::RpcMalformedAuthorizedCredentials, .error = "malformedAuthorizedCredentials", .message = "Malformed authorized credentials."}, + // special system errors {.code = ClioError::RpcInvalidApiVersion, .error = JS(invalid_API_version), .message = "Invalid API version."}, {.code = ClioError::RpcCommandIsMissing, diff --git a/src/rpc/Errors.hpp b/src/rpc/Errors.hpp index 8322cacf4..4ea7a4ebc 100644 --- a/src/rpc/Errors.hpp +++ b/src/rpc/Errors.hpp @@ -39,7 +39,6 @@ enum class ClioError { RpcMalformedRequest = 5001, RpcMalformedOwner = 5002, RpcMalformedAddress = 5003, - RpcInvalidHotWallet = 5004, RpcUnknownOption = 5005, RpcFieldNotFoundTransaction = 5006, RpcMalformedOracleDocumentId = 5007, diff --git a/src/rpc/handlers/AccountTx.hpp b/src/rpc/handlers/AccountTx.hpp index 712ff82cd..0e620a8ca 100644 --- a/src/rpc/handlers/AccountTx.hpp +++ b/src/rpc/handlers/AccountTx.hpp @@ -25,6 +25,7 @@ #include "rpc/common/JsonBool.hpp" #include "rpc/common/MetaProcessors.hpp" #include "rpc/common/Modifiers.hpp" +#include "rpc/common/Specs.hpp" #include "rpc/common/Types.hpp" #include "rpc/common/Validators.hpp" #include "util/TxUtils.hpp" @@ -39,7 +40,6 @@ #include #include -#include #include #include #include @@ -57,8 +57,8 @@ class AccountTxHandler { std::shared_ptr sharedPtrBackend_; public: - // no max limit static constexpr auto kLIMIT_MIN = 1; + static constexpr auto kLIMIT_MAX = 1000; static constexpr auto kLIMIT_DEFAULT = 200; /** @@ -133,7 +133,7 @@ public: {JS(limit), validation::Type{}, validation::Min(1u), - modifiers::Clamp{kLIMIT_MIN, std::numeric_limits::max()}}, + modifiers::Clamp{kLIMIT_MIN, kLIMIT_MAX}}, {JS(marker), meta::WithCustomError{ validation::Type{}, diff --git a/src/rpc/handlers/GatewayBalances.cpp b/src/rpc/handlers/GatewayBalances.cpp index 93f89ddeb..5209f1fa7 100644 --- a/src/rpc/handlers/GatewayBalances.cpp +++ b/src/rpc/handlers/GatewayBalances.cpp @@ -145,10 +145,6 @@ GatewayBalancesHandler::process(GatewayBalancesHandler::Input input, Context con if (auto status = std::get_if(&ret)) return Error{*status}; - auto inHotbalances = [&](auto const& hw) { return output.hotBalances.contains(hw); }; - if (not std::ranges::all_of(input.hotWallets, inHotbalances)) - return Error{Status{ClioError::RpcInvalidHotWallet}}; - output.accountID = input.account; output.ledgerHash = ripple::strHex(lgrInfo.hash); output.ledgerIndex = lgrInfo.seq; diff --git a/src/rpc/handlers/GatewayBalances.hpp b/src/rpc/handlers/GatewayBalances.hpp index 2822c64c7..08a9e97d2 100644 --- a/src/rpc/handlers/GatewayBalances.hpp +++ b/src/rpc/handlers/GatewayBalances.hpp @@ -108,44 +108,51 @@ public: static RpcSpecConstRef spec([[maybe_unused]] uint32_t apiVersion) { - static auto const kHOT_WALLET_VALIDATOR = - validation::CustomValidator{[](boost::json::value const& value, std::string_view key) -> MaybeError { - if (!value.is_string() && !value.is_array()) - return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "NotStringOrArray"}}; + auto const getHotWalletValidator = [](RippledError errCode) { + return validation::CustomValidator{ + [errCode](boost::json::value const& value, std::string_view key) -> MaybeError { + if (!value.is_string() && !value.is_array()) + return Error{Status{errCode, std::string(key) + "NotStringOrArray"}}; - // wallet needs to be an valid accountID or public key - auto const wallets = value.is_array() ? value.as_array() : boost::json::array{value}; - auto const getAccountID = [](auto const& j) -> std::optional { - if (j.is_string()) { - auto const pk = util::parseBase58Wrapper( - ripple::TokenType::AccountPublic, boost::json::value_to(j) - ); + // wallet needs to be an valid accountID or public key + auto const wallets = value.is_array() ? value.as_array() : boost::json::array{value}; + auto const getAccountID = [](auto const& j) -> std::optional { + if (j.is_string()) { + auto const pk = util::parseBase58Wrapper( + ripple::TokenType::AccountPublic, boost::json::value_to(j) + ); - if (pk) - return ripple::calcAccountID(*pk); + if (pk) + return ripple::calcAccountID(*pk); - return util::parseBase58Wrapper(boost::json::value_to(j)); + return util::parseBase58Wrapper(boost::json::value_to(j)); + } + + return {}; + }; + + for (auto const& wallet : wallets) { + if (!getAccountID(wallet)) + return Error{Status{errCode, std::string(key) + "Malformed"}}; } - return {}; - }; - - for (auto const& wallet : wallets) { - if (!getAccountID(wallet)) - return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "Malformed"}}; + return MaybeError{}; } - - return MaybeError{}; - }}; - - static auto const kRPC_SPEC = RpcSpec{ - {JS(account), validation::Required{}, validation::CustomValidators::accountValidator}, - {JS(ledger_hash), validation::CustomValidators::uint256HexStringValidator}, - {JS(ledger_index), validation::CustomValidators::ledgerIndexValidator}, - {JS(hotwallet), kHOT_WALLET_VALIDATOR} + }; }; - return kRPC_SPEC; + static auto const kSPEC_COMMON = RpcSpec{ + {JS(account), validation::Required{}, validation::CustomValidators::accountValidator}, + {JS(ledger_hash), validation::CustomValidators::uint256HexStringValidator}, + {JS(ledger_index), validation::CustomValidators::ledgerIndexValidator} + }; + + auto static const kSPEC_V1 = + RpcSpec{kSPEC_COMMON, {{JS(hotwallet), getHotWalletValidator(ripple::rpcINVALID_HOTWALLET)}}}; + auto static const kSPEC_V2 = + RpcSpec{kSPEC_COMMON, {{JS(hotwallet), getHotWalletValidator(ripple::rpcINVALID_PARAMS)}}}; + + return apiVersion == 1 ? kSPEC_V1 : kSPEC_V2; } /** diff --git a/src/rpc/handlers/LedgerEntry.cpp b/src/rpc/handlers/LedgerEntry.cpp index 36ff61560..c424135aa 100644 --- a/src/rpc/handlers/LedgerEntry.cpp +++ b/src/rpc/handlers/LedgerEntry.cpp @@ -179,6 +179,12 @@ LedgerEntryHandler::process(LedgerEntryHandler::Input input, Context const& ctx) ripple::uint192{std::string_view(boost::json::value_to(input.mptoken->at(JS(mpt_issuance_id)))) }; key = ripple::keylet::mptoken(mptIssuanceID, *holder).key; + } else if (input.permissionedDomain) { + auto const account = ripple::parseBase58( + boost::json::value_to(input.permissionedDomain->at(JS(account))) + ); + auto const seq = input.permissionedDomain->at(JS(seq)).as_int64(); + key = ripple::keylet::permissionedDomain(*account, seq).key; } else { // Must specify 1 of the following fields to indicate what type if (ctx.apiVersion == 1) @@ -313,6 +319,7 @@ tag_invoke(boost::json::value_to_tag, boost::json::va {JS(oracle), ripple::ltORACLE}, {JS(credential), ripple::ltCREDENTIAL}, {JS(mptoken), ripple::ltMPTOKEN}, + {JS(permissioned_domain), ripple::ltPERMISSIONED_DOMAIN} }; auto const parseBridgeFromJson = [](boost::json::value const& bridgeJson) { @@ -399,6 +406,8 @@ tag_invoke(boost::json::value_to_tag, boost::json::va input.credential = parseCredentialFromJson(jv.at(JS(credential))); } else if (jsonObject.contains(JS(mptoken))) { input.mptoken = jv.at(JS(mptoken)).as_object(); + } else if (jsonObject.contains(JS(permissioned_domain))) { + input.permissionedDomain = jv.at(JS(permissioned_domain)).as_object(); } if (jsonObject.contains("include_deleted")) diff --git a/src/rpc/handlers/LedgerEntry.hpp b/src/rpc/handlers/LedgerEntry.hpp index a6ab89ee2..9851ad746 100644 --- a/src/rpc/handlers/LedgerEntry.hpp +++ b/src/rpc/handlers/LedgerEntry.hpp @@ -103,6 +103,7 @@ public: std::optional ticket; std::optional amm; std::optional mptoken; + std::optional permissionedDomain; std::optional bridge; std::optional bridgeAccount; std::optional chainClaimId; @@ -374,6 +375,23 @@ public: }, }, }}, + {JS(permissioned_domain), + meta::WithCustomError{ + validation::Type{}, Status(ClioError::RpcMalformedRequest) + }, + meta::IfType{kMALFORMED_REQUEST_HEX_STRING_VALIDATOR}, + meta::IfType{meta::Section{ + {JS(seq), + meta::WithCustomError{validation::Required{}, Status(ClioError::RpcMalformedRequest)}, + meta::WithCustomError{validation::Type{}, Status(ClioError::RpcMalformedRequest)}}, + { + JS(account), + meta::WithCustomError{validation::Required{}, Status(ClioError::RpcMalformedRequest)}, + meta::WithCustomError{ + validation::CustomValidators::accountBase58Validator, Status(ClioError::RpcMalformedAddress) + }, + }, + }}}, {JS(ledger), check::Deprecated{}}, {"include_deleted", validation::Type{}}, }; diff --git a/src/util/CMakeLists.txt b/src/util/CMakeLists.txt index 8c28fdf95..32e792e80 100644 --- a/src/util/CMakeLists.txt +++ b/src/util/CMakeLists.txt @@ -22,6 +22,7 @@ target_sources( requests/impl/SslContext.cpp ResponseExpirationCache.cpp SignalsHandler.cpp + StopHelper.cpp Taggable.cpp TerminationHandler.cpp TimeUtils.cpp diff --git a/src/util/LedgerUtils.hpp b/src/util/LedgerUtils.hpp index 3ada20a64..2c1916e0d 100644 --- a/src/util/LedgerUtils.hpp +++ b/src/util/LedgerUtils.hpp @@ -117,6 +117,7 @@ class LedgerTypes { LedgerTypeAttribute::chainLedgerType(JS(nunl), ripple::ltNEGATIVE_UNL), LedgerTypeAttribute::deletionBlockerLedgerType(JS(mpt_issuance), ripple::ltMPTOKEN_ISSUANCE), LedgerTypeAttribute::deletionBlockerLedgerType(JS(mptoken), ripple::ltMPTOKEN), + LedgerTypeAttribute::deletionBlockerLedgerType(JS(permissioned_domain), ripple::ltPERMISSIONED_DOMAIN), }; public: diff --git a/src/util/MoveTracker.hpp b/src/util/MoveTracker.hpp new file mode 100644 index 000000000..22d0bd8ae --- /dev/null +++ b/src/util/MoveTracker.hpp @@ -0,0 +1,78 @@ +//------------------------------------------------------------------------------ +/* + This file is part of clio: https://github.com/XRPLF/clio + Copyright (c) 2025, the clio developers. + + Permission to use, copy, modify, and 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. +*/ +//============================================================================== + +#pragma once + +#include + +namespace util { + +/** + * @brief A base-class that can be used to check whether the current instance was moved from + */ +class MoveTracker { + bool wasMoved_ = false; + +protected: + /** + * @brief The function to be used by clients in order to check whether the instance was moved from + * @return true if moved from; false otherwise + */ + [[nodiscard]] bool + wasMoved() const noexcept + { + return wasMoved_; + } + + MoveTracker() = default; // should only be used via inheritance + +public: + virtual ~MoveTracker() = default; + + /** + * @brief Move constructor sets the moved-from state on `other` and resets the state on `this` + * @param other The moved-from object + */ + MoveTracker(MoveTracker&& other) + { + *this = std::move(other); + } + + /** + * @brief Move operator sets the moved-from state on `other` and resets the state on `this` + * @param other The moved-from object + * @return Reference to self + */ + MoveTracker& + operator=(MoveTracker&& other) + { + if (this != &other) { + other.wasMoved_ = true; + wasMoved_ = false; + } + + return *this; + } + + MoveTracker(MoveTracker const&) = default; + MoveTracker& + operator=(MoveTracker const&) = default; +}; + +} // namespace util diff --git a/src/util/Mutex.hpp b/src/util/Mutex.hpp index e507f01fa..bb952f973 100644 --- a/src/util/Mutex.hpp +++ b/src/util/Mutex.hpp @@ -34,7 +34,7 @@ class Mutex; * @tparam LockType type of lock * @tparam MutexType type of mutex */ -template typename LockType, typename MutexType> +template typename LockType, typename MutexType> class Lock { LockType lock_; ProtectedDataType& data_; @@ -129,7 +129,7 @@ public: * @tparam LockType The type of lock to use * @return A lock on the mutex and a reference to the protected data */ - template