ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2026-04-29 15:37:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: Commits for 2.4.0-b3 (#1876)

Browse Source
This commit is contained in:
Alex Kremer
2025-02-06 11:34:21 +00:00
committed by GitHub
parent 1460d590f1 67c989081d
commit 7b043025e8
204 changed files with 6091 additions and 819 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@@ -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**:

2
conanfile.py
View File

@@ -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'
]

11
docker/ci/README.md
View File

@@ -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`).

9
docker/ci/conan/clang.asan Normal file
View File

@@ -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"

9
docker/ci/conan/clang.tsan Normal file
View File

@@ -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"

9
docker/ci/conan/clang.ubsan Normal file
View File

@@ -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"

9
docker/ci/conan/gcc.asan Normal file
View File

@@ -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"

9
docker/ci/conan/gcc.tsan Normal file
View File

@@ -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"

9
docker/ci/conan/gcc.ubsan Normal file
View File

@@ -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"

7
docker/ci/dockerfile
View File

@@ -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

17
docs/build-clio.md
View File

@@ -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
```

30
docs/metrics-and-static-analysis.md
View File

@@ -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
```

12
docs/run-clio.md
View File

@@ -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/).

2
src/app/CMakeLists.txt
View File

@@ -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)

4
src/app/CliArgs.cpp
View File

@@ -47,6 +47,7 @@ CliArgs::parse(int argc, char const* argv[])
("conf,c", po::value<std::string>()->default_value(kDEFAULT_CONFIG_PATH), "configuration file")
("ng-web-server,w", "Use ng-web-server")
("migrate", po::value<std::string>(), "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}
};
}

9
src/app/CliArgs.hpp
View File

@@ -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 <typename ActionType>
requires std::is_same_v<ActionType, Run> or std::is_same_v<ActionType, Exit> or
std::is_same_v<ActionType, Migrate>
std::is_same_v<ActionType, Migrate> or std::is_same_v<ActionType, VerifyConfig>
explicit Action(ActionType&& action) : action_(std::forward<ActionType>(action))
{
}
@@ -86,7 +91,7 @@ public:
}
private:
std::variant<Run, Exit, Migrate> action_;
std::variant<Run, Exit, Migrate, VerifyConfig> action_;
};
/**

17
src/app/ClioApplication.cpp
View File

@@ -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

2
src/app/ClioApplication.hpp
View File

@@ -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:
/**

52
src/app/Stopper.cpp Normal file
View File

@@ -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 <boost/asio/spawn.hpp>
#include <functional>
#include <thread>
#include <utility>
namespace app {
Stopper::~Stopper()
{
if (worker_.joinable())
worker_.join();
}
void
Stopper::setOnStop(std::function<void(boost::asio::yield_context)> 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

118
src/app/Stopper.hpp Normal file
View File

@@ -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 <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <functional>
#include <thread>
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<void(boost::asio::yield_context)> 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<void(boost::asio::yield_context)>
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

58
src/app/VerifyConfig.hpp Normal file
View File

@@ -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 <cstdlib>
#include <iostream>
#include <string_view>
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

2
src/data/AmendmentCenter.hpp
View File

@@ -132,6 +132,8 @@ struct Amendments {
REGISTER(fixAMMv1_2);
REGISTER(AMMClawback);
REGISTER(Credentials);
REGISTER(DynamicNFT);
REGISTER(PermissionedDomains);
// Obsolete but supported by libxrpl
REGISTER(CryptoConditionsSuite);

6
src/data/BackendInterface.hpp
View File

@@ -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
*

47
src/data/CassandraBackend.hpp
View File

@@ -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<uint32_t>(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<uint32_t>(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

18
src/data/DBHelpers.hpp
View File

@@ -107,6 +107,7 @@ struct NFTsData {
ripple::AccountID owner;
std::optional<ripple::Blob> 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)
{
}
};
/**

7
src/data/cassandra/Schema.hpp
View File

@@ -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")
));

9
src/data/cassandra/impl/AsyncExecutor.hpp
View File

@@ -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 <boost/asio.hpp>
@@ -64,8 +65,8 @@ class AsyncExecutor : public std::enable_shared_from_this<AsyncExecutor<Statemen
RetryCallbackType onRetry_;
// does not exist during initial construction, hence optional
std::optional<FutureWithCallbackType> future_;
std::mutex mtx_;
using OptionalFuture = std::optional<FutureWithCallbackType>;
util::Mutex<OptionalFuture> 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<std::scoped_lock>();
future->emplace(handle.asyncExecute(data_, std::move(handler)));
}
};

32
src/data/cassandra/impl/ExecutionStrategy.hpp
View File

@@ -35,6 +35,7 @@
#include <boost/asio/spawn.hpp>
#include <boost/json/object.hpp>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
@@ -192,10 +193,24 @@ public:
template <typename... Args>
void
write(PreparedStatementType const& preparedStatement, Args&&... args)
{
auto statement = preparedStatement.bind(std::forward<Args>(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>(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<StatementType>&& 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.
*

6
src/etl/CacheLoader.hpp
View File

@@ -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();
}
};

2
src/etl/ETLService.cpp
View File

@@ -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;
}

29
src/etl/ETLService.hpp
View File

@@ -42,6 +42,7 @@
#include <org/xrpl/rpc/v1/get_ledger.pb.h>
#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
#include <concepts>
#include <cstddef>
#include <cstdint>
#include <memory>
@@ -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 <typename T>
concept SomeETLService = std::derived_from<T, ETLServiceTag>;
/**
* @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<org::xrpl::rpc::v1::GetLedgerResponse>;
@@ -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();

11
src/etl/LoadBalancer.cpp
View File

@@ -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()
{

24
src/etl/LoadBalancer.hpp
View File

@@ -41,6 +41,7 @@
#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
#include <chrono>
#include <concepts>
#include <cstdint>
#include <expected>
#include <memory>
@@ -51,6 +52,16 @@
namespace etl {
/**
* @brief A tag class to help identify LoadBalancer in templated code.
*/
struct LoadBalancerTag {
virtual ~LoadBalancerTag() = default;
};
template <typename T>
concept SomeLoadBalancer = std::derived_from<T, LoadBalancerTag>;
/**
* @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<ETLState>
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.

16
src/etl/NFTHelpers.cpp
View File

@@ -47,6 +47,17 @@
namespace etl {
std::pair<std::vector<NFTTransactionsData>, std::optional<NFTsData>>
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::vector<NFTTransactionsData>, std::optional<NFTsData>>
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<ripple::STObject>();
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<ripple::STObject>().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 {{}, {}};
}

10
src/etl/NFTHelpers.hpp
View File

@@ -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::vector<NFTTransactionsData>, std::optional<NFTsData>>
getNftokenModifyData(ripple::TxMeta const& txMeta, ripple::STTx const& sttx);
/**
* @brief Get the NFT Token mint data from a transaction
*

9
src/etl/Source.hpp
View File

@@ -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
*

2
src/etl/impl/AmendmentBlockHandler.cpp
View File

@@ -47,7 +47,7 @@ AmendmentBlockHandler::AmendmentBlockHandler(
}
void
AmendmentBlockHandler::onAmendmentBlock()
AmendmentBlockHandler::notifyAmendmentBlocked()
{
state_.get().isAmendmentBlocked = true;
repeat_.start(interval_, action_);

2
src/etl/impl/AmendmentBlockHandler.hpp
View File

@@ -53,7 +53,7 @@ public:
);
void
onAmendmentBlock();
notifyAmendmentBlocked();
};
} // namespace etl::impl

17
src/etl/impl/LedgerLoader.hpp
View File

@@ -57,6 +57,7 @@ struct FormattedTransactionsData {
std::vector<NFTTransactionsData> nfTokenTxData;
std::vector<NFTsData> nfTokensData;
std::vector<MPTHolderData> mptHoldersData;
std::vector<NFTsData> nfTokenURIChanges;
};
namespace etl::impl {
@@ -111,6 +112,7 @@ public:
{
FormattedTransactionsData result;
std::vector<NFTsData> 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;
}

6
src/etl/impl/SourceImpl.hpp
View File

@@ -102,6 +102,12 @@ public:
subscriptionSource_->run();
}
void
stop(boost::asio::yield_context yield) final
{
subscriptionSource_->stop(yield);
}
/**
* @brief Check if source is connected
*

87
src/etl/impl/SubscriptionSource.cpp
View File

@@ -35,7 +35,6 @@
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/use_future.hpp>
#include <boost/beast/http/field.hpp>
#include <boost/json/object.hpp>
#include <boost/json/parse.hpp>
@@ -49,7 +48,6 @@
#include <cstdint>
#include <exception>
#include <expected>
#include <future>
#include <memory>
#include <optional>
#include <stdexcept>
@@ -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<util::requests::RequestError>
@@ -299,6 +282,8 @@ SubscriptionSource::handleError(util::requests::RequestError const& error, boost
logError(error);
if (not stop_) {
retry_.retry([this] { subscribe(); });
} else {
stopHelper_.readyToStop();
}
}

14
src/etl/impl/SubscriptionSource.hpp
View File

@@ -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 <chrono>
#include <cstdint>
#include <functional>
#include <future>
#include <memory>
#include <optional>
#include <string>
@@ -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<util::prometheus::GaugeInt> lastMessageTimeSecondsSinceEpoch_;
std::future<void> 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

2
src/etl/impl/Transformer.hpp
View File

@@ -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};
}

37
src/etlng/AmendmentBlockHandlerInterface.hpp Normal file
View File

@@ -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

5
src/etlng/CMakeLists.txt
View File

@@ -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)

49
src/etlng/LedgerPublisherInterface.hpp Normal file
View File

@@ -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 <chrono>
#include <cstdint>
#include <optional>
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<uint32_t> maxAttempts,
std::chrono::steady_clock::duration attemptsDelay = std::chrono::seconds{1}
) = 0;
};
} // namespace etlng

134
src/etlng/LoadBalancerInterface.hpp Normal file
View File

@@ -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 <boost/asio/spawn.hpp>
#include <boost/json/object.hpp>
#include <boost/json/value.hpp>
#include <org/xrpl/rpc/v1/ledger.pb.h>
#include <xrpl/proto/org/xrpl/rpc/v1/get_ledger.pb.h>
#include <chrono>
#include <cstdint>
#include <expected>
#include <optional>
#include <string>
#include <vector>
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<GetLedgerResponseType>;
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<std::string> The ledger data
*/
virtual std::vector<std::string>
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<std::string> The ledger data
*/
virtual std::vector<std::string>
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<boost::json::object, rpc::ClioError>
forwardToRippled(
boost::json::object const& request,
std::optional<std::string> 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<etl::ETLState>
getETLState() noexcept = 0;
};
} // namespace etlng

52
src/etlng/LoaderInterface.hpp Normal file
View File

@@ -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 <xrpl/protocol/LedgerHeader.h>
#include <optional>
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<ripple::LedgerHeader>
loadInitialLedger(model::LedgerData const& data) = 0;
};
} // namespace etlng

63
src/etlng/Models.hpp
View File

@@ -29,6 +29,7 @@
#include <xrpl/proto/org/xrpl/rpc/v1/ledger.pb.h>
#include <xrpl/protocol/LedgerHeader.h>
#include <xrpl/protocol/STTx.h>
#include <xrpl/protocol/Serializer.h>
#include <xrpl/protocol/TxFormats.h>
#include <xrpl/protocol/TxMeta.h>
@@ -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

42
src/etlng/SchedulerInterface.hpp Normal file
View File

@@ -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 <optional>
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<model::Task>
next() = 0;
};
} // namespace etlng

56
src/etlng/impl/AmendmentBlockHandler.cpp Normal file
View File

@@ -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 <chrono>
#include <functional>
#include <utility>
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

69
src/etlng/impl/AmendmentBlockHandler.hpp Normal file
View File

@@ -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 <boost/asio/io_context.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/steady_timer.hpp>
#include <chrono>
#include <functional>
#include <optional>
namespace etlng::impl {
class AmendmentBlockHandler : public AmendmentBlockHandlerInterface {
public:
using ActionType = std::function<void()>;
private:
std::reference_wrapper<etl::SystemState> state_;
std::chrono::steady_clock::duration interval_;
util::async::AnyExecutionContext ctx_;
std::optional<util::async::AnyOperation<void>> 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

6
src/etlng/impl/AsyncGrpcCall.cpp
View File

@@ -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

2
src/etlng/impl/AsyncGrpcCall.hpp
View File

@@ -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;

4
src/etlng/impl/GrpcSource.cpp
View File

@@ -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;
}

111
src/etlng/impl/Loading.cpp Normal file
View File

@@ -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 <xrpl/protocol/LedgerHeader.h>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <optional>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>
namespace etlng::impl {
Loader::Loader(
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<etl::LedgerFetcherInterface> fetcher,
std::shared_ptr<RegistryInterface> registry,
std::shared_ptr<AmendmentBlockHandlerInterface> 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<std::chrono::duration<double>>([&]() { 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<model::Object> const& data,
std::optional<std::string> 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<ripple::LedgerHeader>
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<std::chrono::seconds>([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

85
src/etlng/impl/Loading.hpp Normal file
View File

@@ -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 <org/xrpl/rpc/v1/ledger.pb.h>
#include <xrpl/basics/Slice.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/strHex.h>
#include <xrpl/proto/org/xrpl/rpc/v1/get_ledger.pb.h>
#include <xrpl/protocol/LedgerHeader.h>
#include <xrpl/protocol/STTx.h>
#include <xrpl/protocol/Serializer.h>
#include <xrpl/protocol/TxMeta.h>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <vector>
namespace etlng::impl {
class Loader : public LoaderInterface, public InitialLoadObserverInterface {
std::shared_ptr<BackendInterface> backend_;
std::shared_ptr<etl::LedgerFetcherInterface> fetcher_;
std::shared_ptr<RegistryInterface> registry_;
std::shared_ptr<AmendmentBlockHandlerInterface> amendmentBlockHandler_;
util::Logger log_{"ETL"};
public:
using RawLedgerObjectType = org::xrpl::rpc::v1::RawLedgerObject;
using GetLedgerResponseType = org::xrpl::rpc::v1::GetLedgerResponse;
using OptionalGetLedgerResponseType = std::optional<GetLedgerResponseType>;
Loader(
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<etl::LedgerFetcherInterface> fetcher,
std::shared_ptr<RegistryInterface> registry,
std::shared_ptr<AmendmentBlockHandlerInterface> amendmentBlockHandler
);
void
load(model::LedgerData const& data) override;
void
onInitialLoadGotMoreObjects(
uint32_t seq,
std::vector<model::Object> const& data,
std::optional<std::string> lastKey
) override;
std::optional<ripple::LedgerHeader>
loadInitialLedger(model::LedgerData const& data) override;
};
} // namespace etlng::impl

151
src/etlng/impl/Scheduling.hpp Normal file
View File

@@ -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 <sys/types.h>
#include <atomic>
#include <cstdint>
#include <functional>
#include <limits>
#include <memory>
#include <optional>
#include <tuple>
#include <type_traits>
#include <utility>
namespace etlng::impl {
template <typename T>
concept SomeScheduler = std::is_base_of_v<SchedulerInterface, std::decay_t<T>>;
class ForwardScheduler : public SchedulerInterface {
std::reference_wrapper<etl::NetworkValidatedLedgersInterface> ledgers_;
uint32_t startSeq_;
std::optional<uint32_t> 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<etl::NetworkValidatedLedgersInterface> ledgers,
uint32_t startSeq,
std::optional<uint32_t> maxSeq = std::nullopt
)
: ledgers_(ledgers), startSeq_(startSeq), maxSeq_(maxSeq), seq_(startSeq)
{
}
[[nodiscard]] std::optional<model::Task>
next() override
{
static constexpr auto kMAX = std::numeric_limits<uint32_t>::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<uint32_t> minSeq = std::nullopt)
: startSeq_(startSeq), minSeq_(minSeq.value_or(0)), seq_(startSeq)
{
}
[[nodiscard]] std::optional<model::Task>
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 <SomeScheduler... Schedulers>
class SchedulerChain : public SchedulerInterface {
std::tuple<Schedulers...> schedulers_;
public:
template <SomeScheduler... Ts>
requires(std::is_same_v<Ts, Schedulers> and ...)
SchedulerChain(Ts&&... schedulers) : schedulers_(std::forward<Ts>(schedulers)...)
{
}
[[nodiscard]] std::optional<model::Task>
next() override
{
std::optional<model::Task> 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<SchedulerChain<std::decay_t<decltype(schedulers)>...>>(
std::forward<decltype(schedulers)>(schedulers)...
);
}
} // namespace etlng::impl

141
src/etlng/impl/TaskManager.cpp Normal file
View File

@@ -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 <chrono>
#include <cstddef>
#include <functional>
#include <ranges>
#include <thread>
#include <utility>
#include <vector>
namespace etlng::impl {
TaskManager::TaskManager(
util::async::AnyExecutionContext&& ctx,
std::reference_wrapper<SchedulerInterface> scheduler,
std::reference_wrapper<ExtractorInterface> extractor,
std::reference_wrapper<LoaderInterface> 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<void>
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<void>
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

94
src/etlng/impl/TaskManager.hpp Normal file
View File

@@ -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 <xrpl/protocol/TxFormats.h>
#include <cstddef>
#include <functional>
#include <vector>
namespace etlng::impl {
class TaskManager {
util::async::AnyExecutionContext ctx_;
std::reference_wrapper<SchedulerInterface> schedulers_;
std::reference_wrapper<ExtractorInterface> extractor_;
std::reference_wrapper<LoaderInterface> loader_;
std::vector<util::async::AnyOperation<void>> extractors_;
std::vector<util::async::AnyOperation<void>> 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<model::LedgerData, ReverseOrderComparator>;
TaskManager(
util::async::AnyExecutionContext&& ctx,
std::reference_wrapper<SchedulerInterface> scheduler,
std::reference_wrapper<ExtractorInterface> extractor,
std::reference_wrapper<LoaderInterface> loader
);
~TaskManager();
void
run(Settings settings);
void
stop();
private:
void
wait();
[[nodiscard]] util::async::AnyOperation<void>
spawnExtractor(util::async::AnyStrand& strand, PriorityQueue& queue);
[[nodiscard]] util::async::AnyOperation<void>
spawnLoader(PriorityQueue& queue);
};
} // namespace etlng::impl

8
src/feed/SubscriptionManager.cpp
View File

@@ -50,7 +50,7 @@ void
SubscriptionManager::pubBookChanges(
ripple::LedgerHeader const& lgrInfo,
std::vector<data::TransactionAndMetadata> 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);
}

17
src/feed/SubscriptionManager.hpp
View File

@@ -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<data::TransactionAndMetadata> 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.

18
src/feed/SubscriptionManagerInterface.hpp
View File

@@ -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<data::TransactionAndMetadata> const& transactions)
const = 0;
pubBookChanges(
ripple::LedgerHeader const& lgrInfo,
std::vector<data::TransactionAndMetadata> 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.

2
src/feed/impl/BookChangesFeed.hpp
View File

@@ -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<data::TransactionAndMetadata> const& transactions) const
pub(ripple::LedgerHeader const& lgrInfo, std::vector<data::TransactionAndMetadata> const& transactions)
{
SingleFeedBase::pub(boost::json::serialize(rpc::computeBookChanges(lgrInfo, transactions)));
}

2
src/feed/impl/ForwardFeed.hpp
View File

@@ -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));
}

2
src/feed/impl/LedgerFeed.cpp
View File

@@ -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)));
}

2
src/feed/impl/LedgerFeed.hpp
View File

@@ -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

2
src/feed/impl/SingleFeedBase.cpp
View File

@@ -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<std::string>(msg);

2
src/feed/impl/SingleFeedBase.hpp
View File

@@ -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.

22
src/feed/impl/TrackableSignal.hpp
View File

@@ -19,6 +19,8 @@
#pragma once
#include "util/Mutex.hpp"
#include <boost/signals2.hpp>
#include <boost/signals2/connection.hpp>
#include <boost/signals2/variadic_signal.hpp>
@@ -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<ConnectionPtr, boost::signals2::connection> connections_;
mutable std::mutex mutex_;
using ConnectionsMap = std::unordered_map<ConnectionPtr, boost::signals2::connection>;
util::Mutex<ConnectionsMap> connections_;
using SignalType = boost::signals2::signal<void(Args...)>;
SignalType signal_;
@@ -64,8 +66,8 @@ public:
bool
connectTrackableSlot(ConnectionSharedPtr const& trackable, std::function<void(Args...)> slot)
{
std::scoped_lock const lk(mutex_);
if (connections_.contains(trackable.get())) {
auto connections = connections_.template lock<std::scoped_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<std::scoped_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<std::scoped_lock>()->size();
}
};
} // namespace feed::impl

25
src/feed/impl/TrackableSignalMap.hpp
View File

@@ -20,6 +20,7 @@
#pragma once
#include "feed/impl/TrackableSignal.hpp"
#include "util/Mutex.hpp"
#include <boost/signals2.hpp>
@@ -49,8 +50,8 @@ class TrackableSignalMap {
using ConnectionPtr = Session*;
using ConnectionSharedPtr = std::shared_ptr<Session>;
mutable std::mutex mutex_;
std::unordered_map<Key, TrackableSignal<Session, Args...>> signalsMap_;
using SignalsMap = std::unordered_map<Key, TrackableSignal<Session, Args...>>;
util::Mutex<SignalsMap> signalsMap_;
public:
/**
@@ -66,8 +67,8 @@ public:
bool
connectTrackableSlot(ConnectionSharedPtr const& trackable, Key const& key, std::function<void(Args...)> slot)
{
std::scoped_lock const lk(mutex_);
return signalsMap_[key].connectTrackableSlot(trackable, slot);
auto map = signalsMap_.template lock<std::scoped_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<std::scoped_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<std::scoped_lock>();
if (map->contains(key))
map->operator[](key).emit(args...);
}
};
} // namespace feed::impl

36
src/main/Main.cpp
View File

@@ -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 <cstdlib>
#include <exception>
@@ -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();

2
src/migration/CMakeLists.txt
View File

@@ -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)

2
src/migration/MigrationApplication.cpp
View File

@@ -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

2
src/migration/MigrationApplication.hpp
View File

@@ -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

65
src/migration/MigrationInspectorFactory.hpp Normal file
View File

@@ -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 <boost/algorithm/string.hpp>
#include <boost/algorithm/string/predicate.hpp>
#include <memory>
#include <utility>
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<MigrationInspectorInterface> instance
*/
inline std::shared_ptr<MigrationInspectorInterface>
makeMigrationInspector(
util::config::ClioConfigDefinition const& config,
std::shared_ptr<BackendInterface> const& backend
)
{
ASSERT(backend != nullptr, "Backend is not initialized");
auto inspector = std::make_shared<migration::cassandra::CassandraMigrationInspector>(backend);
// Database is empty, we need to initialize the migration table if it is a writeable backend
if (not config.get<bool>("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

79
src/migration/MigrationInspectorInterface.hpp Normal file
View File

@@ -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 <string>
#include <tuple>
#include <vector>
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<std::tuple<std::string, MigratorStatus>>
allMigratorsStatusPairs() const = 0;
/**
* @brief Get all registered migrators' names
*
* @return A vector of migrators' names
*/
virtual std::vector<std::string>
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

48
src/migration/MigrationManagerInterface.hpp
View File

@@ -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 <string>
#include <tuple>
#include <vector>
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<std::tuple<std::string, MigratorStatus>>
allMigratorsStatusPairs() const = 0;
/**
* @brief Get all registered migrators' names
*
* @return A vector of migrators' names
*/
virtual std::vector<std::string>
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

2
src/migration/MigratiorStatus.hpp
View File

@@ -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

2
src/migration/MigratorStatus.cpp
View File

@@ -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

10
src/migration/README.md
View File

@@ -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.

2
src/migration/cassandra/CassandraMigrationBackend.hpp
View File

@@ -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
src/migration/cassandra/CassandraMigrationManager.hpp
View File

@@ -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<BackendType, ExampleMigrator>
template <typename BackendType>
using CassandraSupportedMigrators = migration::impl::MigratorsRegister<BackendType>;
// Register with MigrationBackend which proceeds the migration
using MigrationProcesser = CassandraSupportedMigrators<CassandraMigrationBackend>;
// Instantiates with the backend which supports actual migration running
using MigrationProcesser = CassandraSupportedMigrators<migration::cassandra::CassandraMigrationBackend>;
// Instantiates with backend interface, it doesn't support actual migration. But it can be used to inspect the migrators
// status
using MigrationQuerier = CassandraSupportedMigrators<data::BackendInterface>;
} // namespace
namespace migration::cassandra {
using CassandraMigrationInspector = migration::impl::MigrationInspectorBase<MigrationQuerier>;
// The Cassandra migration manager
using CassandraMigrationManager = migration::impl::MigrationManagerBase<MigrationProcesser>;
} // namespace migration::cassandra

2
src/migration/cassandra/impl/CassandraMigrationSchema.hpp
View File

@@ -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

2
src/migration/cassandra/impl/FullTableScanner.hpp
View File

@@ -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

2
src/migration/cassandra/impl/FullTableScannerAdapterBase.hpp
View File

@@ -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

2
src/migration/cassandra/impl/ObjectsAdapter.cpp
View File

@@ -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

2
src/migration/cassandra/impl/ObjectsAdapter.hpp
View File

@@ -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

2
src/migration/cassandra/impl/Spec.hpp
View File

@@ -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

2
src/migration/cassandra/impl/TransactionsAdapter.cpp
View File

@@ -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

2
src/migration/cassandra/impl/TransactionsAdapter.hpp
View File

@@ -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

2
src/migration/cassandra/impl/Types.hpp
View File

@@ -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

121
src/migration/impl/MigrationInspectorBase.hpp Normal file
View File

@@ -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 <memory>
#include <ranges>
#include <string>
#include <tuple>
#include <vector>
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 <typename SupportedMigrators>
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<typename SupportedMigrators::BackendType> 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<std::tuple<std::string, MigratorStatus>>
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<std::string>
allMigratorsNames() const override
{
auto const names = migrators_.getMigratorNames();
return std::vector<std::string>{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

61
src/migration/impl/MigrationManagerBase.hpp
View File

@@ -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 <memory>
#include <string>
#include <tuple>
#include <utility>
#include <vector>
namespace migration::impl {
@@ -38,8 +36,7 @@ namespace migration::impl {
* @tparam SupportedMigrators The migrators resgister that contains all the migrators
*/
template <typename SupportedMigrators>
class MigrationManagerBase : public MigrationManagerInterface {
SupportedMigrators migrators_;
class MigrationManagerBase : public MigrationManagerInterface, public MigrationInspectorBase<SupportedMigrators> {
// contains only migration related settings
util::config::ObjectView config_;
@@ -54,7 +51,7 @@ public:
std::shared_ptr<typename SupportedMigrators::BackendType> backend,
util::config::ObjectView config
)
: migrators_{backend}, config_{std::move(config)}
: MigrationInspectorBase<SupportedMigrators>{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<std::tuple<std::string, MigratorStatus>>
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<std::string>
allMigratorsNames() const override
{
auto const names = migrators_.getMigratorNames();
return std::vector<std::string>{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_);
}
};

2
src/migration/impl/MigrationManagerFactory.cpp
View File

@@ -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

2
src/migration/impl/MigrationManagerFactory.hpp
View File

@@ -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

48
src/migration/impl/MigratorsRegister.hpp
View File

@@ -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 <array>
#include <iterator>
#include <memory>
#include <optional>
#include <ranges>
#include <string>
#include <string_view>
#include <tuple>
#include <utility>
#include <vector>
namespace migration::impl {
@@ -47,6 +50,11 @@ concept MigrationBackend = requires { requires std::same_as<typename MigratorTyp
template <typename Backend, typename... MigratorType>
concept BackendMatchAllMigrators = (MigrationBackend<Backend, MigratorType> && ...);
template <typename T>
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 <typename First, typename... Rest>
static constexpr bool
canBlockClioHelper(std::string_view targetName)
{
if (targetName == First::kNAME) {
if constexpr (HasCanBlockClio<First>) {
return First::kCAN_BLOCK_CLIO;
}
return false;
}
if constexpr (sizeof...(Rest) > 0) {
return canBlockClioHelper<Rest...>(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<bool>
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<MigratorType...>(name);
}
}
};
} // namespace migration::impl

2
src/migration/impl/Spec.hpp
View File

@@ -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

2
src/rpc/Errors.cpp
View File

@@ -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,

1
src/rpc/Errors.hpp
View File

@@ -39,7 +39,6 @@ enum class ClioError {
RpcMalformedRequest = 5001,
RpcMalformedOwner = 5002,
RpcMalformedAddress = 5003,
RpcInvalidHotWallet = 5004,
RpcUnknownOption = 5005,
RpcFieldNotFoundTransaction = 5006,
RpcMalformedOracleDocumentId = 5007,

Some files were not shown because too many files have changed in this diff Show More