20 #include <ripple/app/ledger/InboundLedgers.h>
21 #include <ripple/app/ledger/LedgerMaster.h>
22 #include <ripple/app/misc/NetworkOPs.h>
23 #include <ripple/basics/ByteUtilities.h>
24 #include <ripple/basics/chrono.h>
25 #include <ripple/basics/random.h>
26 #include <ripple/core/ConfigSections.h>
27 #include <ripple/nodestore/DummyScheduler.h>
28 #include <ripple/nodestore/impl/DatabaseShardImp.h>
29 #include <ripple/overlay/Overlay.h>
30 #include <ripple/overlay/predicates.h>
31 #include <ripple/protocol/HashPrefix.h>
33 #include <boost/algorithm/string/predicate.hpp>
36 #include <sys/statvfs.h>
60 , avgShardFileSz_(ledgersPerShard_ *
kilobytes(192ull))
73 JLOG(
j_.
error()) <<
"already initialized";
79 JLOG(
j_.
error()) <<
"invalid configuration file settings";
85 using namespace boost::filesystem;
92 for (
auto const& path : paths)
96 if (!is_directory(path))
98 JLOG(
j_.
error()) << path <<
" must be a directory";
102 else if (!create_directories(path))
105 <<
"failed to create path: " + path.string();
117 ctx_ = std::make_unique<nudb::context>();
122 for (
auto const& path : paths)
124 for (
auto const& it : directory_iterator(path))
127 if (!is_directory(it))
131 auto const shardDir{it.path()};
132 auto dirName{shardDir.stem().string()};
134 dirName.begin(), dirName.end(), [](
auto c) {
135 return ::isdigit(static_cast<unsigned char>(c));
146 <<
"shard " << shardIndex
147 <<
" ignored, comes before earliest shard index "
156 <<
"shard " << shardIndex
157 <<
" previously failed import, removing";
158 remove_all(shardDir);
162 auto shard{std::make_shared<Shard>(
163 app_, *
this, shardIndex, shardDir.parent_path(),
j_)};
167 shard->removeOnDestroy();
169 <<
"shard " << shardIndex <<
" removed, "
170 << (shard->isLegacy() ?
"legacy" :
"corrupted")
175 switch (shard->getState())
180 shards_.emplace(shardIndex, std::move(shard));
185 shards_.emplace(shardIndex, std::move(shard))
195 <<
"more than one shard being acquired";
199 shards_.emplace(shardIndex, std::move(shard));
205 <<
"shard " << shardIndex <<
" invalid state";
213 JLOG(
j_.
fatal()) <<
"Exception caught in function " << __func__
214 <<
". Error: " << e.
what();
227 boost::optional<std::uint32_t>
230 boost::optional<std::uint32_t> shardIndex;
239 return it->second->prepare();
254 JLOG(
j_.
debug()) <<
"no new shards to add";
262 auto const pathDesignation = [
this, shardIndex = *shardIndex]() {
267 if (!pathDesignation)
270 auto const needsHistoricalPath =
273 auto shard = [
this, shardIndex, needsHistoricalPath] {
275 return std::make_unique<Shard>(
286 auto const ledgerSeq{shard->prepare()};
289 shards_.emplace(*shardIndex, std::move(shard));
298 auto fail = [j =
j_, &shardIndexes](
300 boost::optional<std::uint32_t> shardIndex = boost::none) {
301 auto multipleIndexPrequel = [&shardIndexes] {
304 shardIndexes.
begin(),
306 indexesAsString.
begin(),
307 [](uint32_t
const index) { return std::to_string(index); });
310 (shardIndexes.
size() > 1 ?
"s " :
" ") +
311 boost::algorithm::join(indexesAsString,
", ");
316 : multipleIndexPrequel();
318 JLOG(j.error()) << prequel <<
" " << msg;
326 return fail(
"cannot be stored at this time");
328 auto historicalShardsToPrepare = 0;
330 for (
auto const shardIndex : shardIndexes)
335 "comes before earliest shard index " +
346 return fail(
"invalid index", shardIndex);
353 return fail(
"invalid index", shardIndex);
357 return fail(
"is already stored", shardIndex);
360 return fail(
"is already queued for import", shardIndex);
365 ++historicalShardsToPrepare;
372 return fail(
"maximum number of historical shards reached");
374 if (historicalShardsToPrepare)
379 return fail(
"insufficient storage space available");
382 if (
auto const recentShardsToPrepare =
383 shardIndexes.size() - historicalShardsToPrepare;
384 recentShardsToPrepare)
389 return fail(
"insufficient storage space available");
392 for (
auto const shardIndex : shardIndexes)
394 auto const prepareSuccessful =
397 (void)prepareSuccessful;
398 assert(prepareSuccessful);
422 rs.insert(shardIndex);
434 boost::filesystem::path
const& srcDir)
438 JLOG(
j_.
error()) <<
"shard " << shardIndex <<
" " << msg;
445 using namespace boost::filesystem;
448 if (!is_directory(srcDir) || is_empty(srcDir))
451 "invalid source directory " + srcDir.string(),
458 std::string(
". Exception caught in function ") + __func__ +
459 ". Error: " + e.
what(),
472 return fail(
"already exists", lock);
476 return fail(
"was not prepared for import", lock);
478 auto const pathDesignation{
480 if (!pathDesignation)
481 return fail(
"failed to import", lock);
490 auto renameDir = [&](path
const& src, path
const& dst) {
498 std::string(
". Exception caught in function ") + __func__ +
499 ". Error: " + e.
what(),
506 if (!renameDir(srcDir, dstDir))
510 auto shard{std::make_unique<Shard>(
511 app_, *
this, shardIndex, dstDir.parent_path(),
j_)};
516 renameDir(dstDir, srcDir);
520 auto const [it, inserted] = [&]() {
523 return shards_.emplace(shardIndex, std::move(shard));
529 renameDir(dstDir, srcDir);
547 auto const it{
shards_.find(shardIndex)};
554 switch (shard->getState())
559 if (shard->containsLedger(ledgerSeq))
572 JLOG(
j_.
error()) <<
"shard " << shardIndex <<
" " << msg;
576 auto ledger{std::make_shared<Ledger>(
581 if (ledger->info().seq != ledgerSeq)
584 "encountered invalid ledger sequence " +
std::to_string(ledgerSeq));
586 if (ledger->info().hash != hash)
589 "encountered invalid ledger hash " +
to_string(hash) +
594 if (!ledger->stateMap().fetchRoot(
595 SHAMapHash{ledger->info().accountHash},
nullptr))
598 "is missing root STATE node on hash " +
to_string(hash) +
602 if (ledger->info().txHash.isNonZero())
604 if (!ledger->txMap().fetchRoot(
608 "is missing root TXN node on hash " +
to_string(hash) +
618 auto const ledgerSeq{ledger->info().seq};
619 if (ledger->info().hash.isZero())
621 JLOG(
j_.
error()) <<
"zero ledger hash for ledger sequence "
625 if (ledger->info().accountHash.isZero())
627 JLOG(
j_.
error()) <<
"zero account hash for ledger sequence "
631 if (ledger->stateMap().getHash().isNonZero() &&
632 !ledger->stateMap().isValid())
634 JLOG(
j_.
error()) <<
"invalid state map for ledger sequence "
638 if (ledger->info().txHash.isNonZero() && !ledger->txMap().isValid())
640 JLOG(
j_.
error()) <<
"invalid transaction map for ledger sequence "
654 <<
"shard " << shardIndex <<
" is not being acquired";
658 auto const it{
shards_.find(shardIndex)};
662 <<
"shard " << shardIndex <<
" is not being acquired";
668 if (shard->containsLedger(ledgerSeq))
670 JLOG(
j_.
trace()) <<
"shard " << shardIndex <<
" ledger already stored";
713 for (
auto const& e : shards)
718 if (
auto const shard{e.lock()}; shard)
721 JLOG(
j_.
warn()) <<
" shard " << shardIndex <<
" unexpired";
739 JLOG(
j_.
error()) <<
"invalid source database";
746 auto loadLedger = [&](
bool ascendSort =
747 true) -> boost::optional<std::uint32_t> {
751 "WHERE LedgerSeq >= " +
753 " order by LedgerSeq " + (ascendSort ?
"asc" :
"desc") +
757 if (!ledger || ledgerSeq == 0)
759 JLOG(
j_.
error()) <<
"no suitable ledgers were found in"
760 " the SQLite database to import";
767 auto ledgerSeq{loadLedger()};
777 ledgerSeq = loadLedger(
false);
786 if (latestIndex < earliestIndex)
788 JLOG(
j_.
error()) <<
"no suitable ledgers were found in"
789 " the SQLite database to import";
798 shardIndex <= latestIndex;
801 auto const pathDesignation =
804 if (!pathDesignation)
807 auto const needsHistoricalPath =
814 <<
"shard " << shardIndex <<
" already being acquired";
822 <<
"shard " << shardIndex <<
" already being imported";
829 JLOG(
j_.
debug()) <<
"shard " << shardIndex <<
" already stored";
838 auto const numLedgers{
842 if (ledgerHashes.size() != numLedgers)
850 JLOG(
j_.
warn()) <<
"SQLite ledger sequence " << n
851 <<
" mismatches node store";
865 std::make_unique<Shard>(
app_, *
this, shardIndex, path,
j_)};
876 JLOG(
j_.
error()) <<
"shard " << shardIndex
877 <<
" failed to create temp marker file";
878 shard->removeOnDestroy();
886 boost::optional<uint256> lastLedgerHash;
888 while (
auto const ledgerSeq = shard->prepare())
891 if (!ledger || ledger->info().seq != ledgerSeq)
894 auto const result{shard->storeLedger(ledger, recentStored)};
899 if (!shard->setLedgerStored(ledger))
902 if (!lastLedgerHash && ledgerSeq ==
lastLedgerSeq(shardIndex))
903 lastLedgerHash = ledger->info().hash;
905 recentStored = std::move(ledger);
908 using namespace boost::filesystem;
921 if (shard->storeNodeObject(nodeObject))
927 remove_all(markerFile);
929 JLOG(
j_.
debug()) <<
"shard " << shardIndex
930 <<
" was successfully imported";
932 shards_.emplace(shardIndex, std::move(shard))
943 JLOG(
j_.
fatal()) <<
"shard index " << shardIndex
944 <<
". Exception caught in function "
945 << __func__ <<
". Error: " << e.
what();
953 <<
"shard " << shardIndex <<
" failed to import";
954 shard->removeOnDestroy();
978 return shard->getWriteLoad();
995 <<
"shard " << shardIndex <<
" is not being acquired";
999 auto const it{
shards_.find(shardIndex)};
1003 <<
"shard " << shardIndex <<
" is not being acquired";
1009 auto const nodeObject{
1011 if (shard->storeNodeObject(nodeObject))
1018 auto const ledgerSeq{srcLedger->info().seq};
1028 <<
"shard " << shardIndex <<
" is not being acquired";
1032 auto const it{
shards_.find(shardIndex)};
1036 <<
"shard " << shardIndex <<
" is not being acquired";
1042 auto const result{shard->storeLedger(srcLedger,
nullptr)};
1044 if (result.error || result.count == 0 || result.size == 0)
1066 for (
auto const& e : shards)
1068 if (
auto const shard{e.lock()}; shard && shard->isOpen())
1079 JLOG(
j_.
trace()) <<
"Open shards exceed configured limit of "
1090 return lhsShard->getLastUse() < rhsShard->getLastUse();
1093 for (
auto it{openFinals.
cbegin()};
1096 if ((*it)->tryClose())
1097 it = openFinals.
erase(it);
1124 get_if_exists<std::uint32_t>(
1125 section,
"earliest_seq", shardDBEarliestSeq);
1128 get_if_exists<std::uint32_t>(
1133 if (shardDBEarliestSeq != nodeDBEarliestSeq)
1137 "] define different 'earliest_seq' values");
1141 using namespace boost::filesystem;
1142 if (!get_if_exists<path>(section,
"path",
dir_))
1143 return fail(
"'path' missing");
1148 Section const& historicalShardPaths =
1149 config.section(SECTION_HISTORICAL_SHARD_PATHS);
1151 auto values = historicalShardPaths.
values();
1153 std::sort(values.begin(), values.end());
1154 values.erase(
std::unique(values.begin(), values.end()), values.end());
1156 for (
auto const& s : values)
1158 auto const dir = path(s);
1162 "the 'path' cannot also be in the "
1163 "'historical_shard_path' section");
1170 if (section.exists(
"ledgers_per_shard"))
1173 if (!config.standalone())
1174 return fail(
"'ledgers_per_shard' only honored in stand alone");
1178 return fail(
"'ledgers_per_shard' must be a multiple of 256");
1185 backendName_ = get<std::string>(section,
"type",
"nudb");
1187 return fail(
"'type' value unsupported");
1202 auto const it{
shards_.find(shardIndex)};
1208 return shard->fetchNodeObject(hash, fetchReport);
1211 boost::optional<std::uint32_t>
1219 auto const maxShardIndex{[
this, validLedgerSeq]() {
1228 if (
shards_.size() >= maxNumShards)
1231 if (maxShardIndex < 1024 ||
1232 static_cast<float>(
shards_.size()) / maxNumShards > 0.5f)
1240 shardIndex <= maxShardIndex;
1263 for (
int i = 0; i < 40; ++i)
1281 boost::optional<uint256>
const& expectedHash)
1290 auto shard{wptr.lock()};
1293 JLOG(
j_.
debug()) <<
"Shard removed before being finalized";
1297 if (!shard->finalize(writeSQLite, expectedHash))
1316 if (shard->index() < boundaryIndex)
1320 shard->getDir().parent_path() ==
dir_)
1323 JLOG(
j_.
warn()) <<
"shard " << shard->index()
1324 <<
" is not stored at a historical path";
1332 assert(!boundaryIndex || shard->index() - boundaryIndex <= 1);
1334 auto& recentShard = shard->index() == boundaryIndex
1339 recentShard = shard->index();
1341 if (shard->getDir().parent_path() !=
dir_)
1343 JLOG(
j_.
warn()) <<
"shard " << shard->index()
1344 <<
" is not stored at the path";
1355 protocol::TMPeerShardInfo message;
1357 message.set_nodepubkey(publicKey.data(), publicKey.size());
1360 message, protocol::mtPEER_SHARD_INFO)));
1382 for (
auto const& e : shards)
1384 if (
auto const shard{e.lock()}; shard)
1386 auto const [sz, fd] = shard->getFileInfo();
1409 JLOG(
j_.
warn()) <<
"maximum number of historical shards reached";
1420 <<
"maximum shard store size exceeds available storage space";
1434 rs.insert(e.second->index());
1459 auto const availableSpace =
1460 boost::filesystem::space(path).available;
1480 if (numShards <= shardCap)
1483 numShards -= shardCap;
1488 JLOG(
j_.
fatal()) <<
"Exception caught in function " << __func__
1489 <<
". Error: " << e.
what();
1501 if (!shard->setLedgerStored(ledger))
1511 if (
auto const it{
shards_.find(shard->index())}; it !=
shards_.end())
1521 <<
"shard " << shard->index() <<
" is no longer being acquired";
1544 if ((
shards_.erase(shard->index()) > 0) &&
1551 shard->removeOnDestroy();
1581 shards_.begin(),
shards_.end(), [boundaryIndex](
auto const& entry) {
1582 return entry.first < boundaryIndex;
1593 auto const latestShardIndex =
1598 auto const removeShard =
1608 JLOG(
j_.
warn()) <<
"can't find shard to remove";
1613 JLOG(
j_.
warn()) <<
"can't find shard to remove";
1617 auto const keepShard =
1618 [
this, &lock, removeShard, separateHistoricalPath](
1623 <<
"maximum number of historical shards reached";
1625 removeShard(shardIndex);
1628 if (separateHistoricalPath &&
1631 JLOG(
j_.
error()) <<
"insufficient storage space available";
1633 removeShard(shardIndex);
1642 auto const moveShard = [
this,
1648 auto& shard{it->second};
1653 if (!shard->tryClose())
1656 <<
"can't close shard to move to historical path";
1663 boost::filesystem::rename(
1664 shard->getDir().string(),
1669 JLOG(
j_.
error()) <<
"shard " << shardIndex
1670 <<
" failed to move to historical storage";
1676 std::make_shared<Shard>(
app_, *
this, shardIndex, dst,
j_);
1681 JLOG(
j_.
error()) <<
"shard " << shardIndex
1682 <<
" failed to open in historical storage";
1683 shard->removeOnDestroy();
1690 <<
"can't find shard to move to historical path";
1695 bool const curNotSynched =
1702 if (curNotSynched || prevNotSynched)
1707 if (keepShard(*prev) && separateHistoricalPath)
1718 if (cur == latestShardIndex - 1)
1727 if (keepShard(*cur) && separateHistoricalPath)
1747 auto const isHistoricalShard = shardIndex < boundaryIndex;
1756 JLOG(
j_.
error()) <<
"maximum number of historical shards reached";
1762 JLOG(
j_.
error()) <<
"insufficient storage space available";
1770 boost::filesystem::path
1778 boost::filesystem::path historicalShardPath;
1787 if (potentialPaths.
empty())
1789 JLOG(
j_.
error()) <<
"failed to select a historical shard path";
1794 potentialPaths.
begin(),
1795 potentialPaths.
end(),
1796 &historicalShardPath,
1800 return historicalShardPath;
1815 struct statvfs buffer;
1816 if (statvfs(path.c_str(), &buffer))
1819 <<
"failed to acquire stats for 'historical_shard_path': "
1824 filesystemIDs[buffer.f_fsid].push_back(path.string());
1828 for (
auto const& entry : filesystemIDs)
1831 if (entry.second.size() > 1)
1836 <<
"The following paths correspond to the same filesystem: "
1837 << boost::algorithm::join(entry.second,
", ")
1838 <<
". Each configured historical storage path should"
1839 " be on a unique device or filesystem.";
1860 uniqueCapacities[boost::filesystem::space(path).available].push_back(
1863 for (
auto const& entry : uniqueCapacities)
1866 if (entry.second.size() > 1)
1871 <<
"Each of the following paths have " << entry.first
1872 <<
" bytes free, and may be located on the same device"
1874 << boost::algorithm::join(entry.second,
", ")
1875 <<
". Each configured historical storage path should"
1876 " be on a unique device or file system.";
1897 if (section.empty())
1900 return std::make_unique<DatabaseShardImp>(
1901 app, parent,
"ShardStore", scheduler, readThreads, j);