external/ton
1
0
Fork 0
mirror of https://github.com/ton-blockchain/ton synced 2025-03-09 15:40:10 +00:00
Code Issues Releases Wiki Activity

Accelerator: partial fullnodes (#1393)

Browse source 
* Accelerator: partial fullnodes

1) Node can monitor a subset of shards
2) New archive slice format (sharded)
3) Validators are still required to have all shards
4) Support partial liteservers in lite-client, blockchain explorer, tonlib
5) Proxy liteserver

* Fix compilation error
This commit is contained in:
SpyCheese 2024-11-26 15:46:58 +04:00 • committed by GitHub
parent 62444100f5
commit 954a96a077
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
83 changed files with 3213 additions and 1113 deletions
Download patch file Download diff file Expand all files Collapse all files

296
validator/manager.cpp
View file

@ -20,11 +20,9 @@
#include "checksum.h"
#include "td/utils/buffer.h"
#include "validator-group.hpp"
#include "adnl/utils.hpp"
#include "downloaders/wait-block-state.hpp"
#include "downloaders/wait-block-state-merge.hpp"
#include "downloaders/wait-block-data.hpp"
#include "validator-group.hpp"
#include "fabric.h"
#include "manager.h"
#include "validate-broadcast.hpp"
@ -202,7 +200,7 @@ void ValidatorManagerImpl::validate_block(ReceivedBlock block, td::Promise<Block
CHECK(blkid.is_masterchain());
auto P = td::PromiseCreator::lambda(
[SelfId = actor_id(this), promise = std::move(promise), id = block.id](td::Result<td::Unit> R) mutable {
[SelfId = actor_id(this), promise = std::move(promise), id = blkid](td::Result<td::Unit> R) mutable {
if (R.is_error()) {
promise.set_error(R.move_as_error());
} else {
@ -217,12 +215,26 @@ void ValidatorManagerImpl::prevalidate_block(BlockBroadcast broadcast, td::Promi
promise.set_error(td::Status::Error(ErrorCode::notready, "node not started"));
return;
}
if (!need_monitor(broadcast.block_id.shard_full())) {
promise.set_error(td::Status::Error("not monitoring shard"));
return;
}
promise = [SelfId = actor_id(this), promise = std::move(promise), block_id = broadcast.block_id,
cc_seqno = broadcast.catchain_seqno](td::Result<td::Unit> R) mutable {
if (R.is_ok()) {
td::actor::send_closure(SelfId, &ValidatorManagerImpl::validated_block_broadcast, block_id, cc_seqno);
}
promise.set_result(std::move(R));
};
td::actor::create_actor<ValidateBroadcast>("broadcast", std::move(broadcast), last_masterchain_block_handle_,
last_masterchain_state_, last_known_key_block_handle_, actor_id(this),
td::Timestamp::in(2.0), std::move(promise))
.release();
}
void ValidatorManagerImpl::validated_block_broadcast(BlockIdExt block_id, CatchainSeqno cc_seqno) {
}
void ValidatorManagerImpl::sync_complete(td::Promise<td::Unit> promise) {
started_ = true;
@ -474,7 +486,7 @@ void ValidatorManagerImpl::new_ihr_message(td::BufferSlice data) {
}
void ValidatorManagerImpl::new_shard_block(BlockIdExt block_id, CatchainSeqno cc_seqno, td::BufferSlice data) {
if (!is_validator()) {
if (!is_validator() && !cached_block_candidates_.count(block_id)) {
return;
}
if (!last_masterchain_block_handle_) {
@ -508,37 +520,36 @@ void ValidatorManagerImpl::new_block_candidate(BlockIdExt block_id, td::BufferSl
if (!started_) {
return;
}
if (!need_monitor(block_id.shard_full())) {
VLOG(VALIDATOR_DEBUG) << "dropping block candidate broadcast: not monitoring shard";
return;
}
add_cached_block_candidate(ReceivedBlock{block_id, std::move(data)});
}
void ValidatorManagerImpl::add_shard_block_description(td::Ref<ShardTopBlockDescription> desc) {
if (desc->may_be_valid(last_masterchain_block_handle_, last_masterchain_state_)) {
auto it = shard_blocks_.find(ShardTopBlockDescriptionId{desc->shard(), desc->catchain_seqno()});
if (it != shard_blocks_.end() && desc->block_id().id.seqno <= it->second->block_id().id.seqno) {
VLOG(VALIDATOR_DEBUG) << "dropping duplicate shard block broadcast";
return;
}
shard_blocks_[ShardTopBlockDescriptionId{desc->block_id().shard_full(), desc->catchain_seqno()}] = desc;
VLOG(VALIDATOR_DEBUG) << "new shard block descr for " << desc->block_id();
if (last_masterchain_block_handle_ && last_masterchain_seqno_ > 0 &&
desc->generated_at() < last_masterchain_block_handle_->unix_time() + 60) {
delay_action(
[SelfId = actor_id(this), desc]() {
auto P = td::PromiseCreator::lambda([](td::Result<td::Ref<ShardState>> R) {
if (R.is_error()) {
auto S = R.move_as_error();
if (S.code() != ErrorCode::timeout && S.code() != ErrorCode::notready) {
VLOG(VALIDATOR_NOTICE) << "failed to get shard state: " << S;
} else {
VLOG(VALIDATOR_DEBUG) << "failed to get shard state: " << S;
}
}
});
td::actor::send_closure(SelfId, &ValidatorManager::wait_block_state_short, desc->block_id(), 0,
td::Timestamp::in(60.0), std::move(P));
},
td::Timestamp::in(1.0));
}
if (!desc->may_be_valid(last_masterchain_block_handle_, last_masterchain_state_)) {
return;
}
auto it = shard_blocks_.find(ShardTopBlockDescriptionId{desc->shard(), desc->catchain_seqno()});
if (it != shard_blocks_.end() && desc->block_id().id.seqno <= it->second->block_id().id.seqno) {
VLOG(VALIDATOR_DEBUG) << "dropping duplicate shard block broadcast";
return;
}
shard_blocks_[ShardTopBlockDescriptionId{desc->block_id().shard_full(), desc->catchain_seqno()}] = desc;
VLOG(VALIDATOR_DEBUG) << "new shard block descr for " << desc->block_id();
if (need_monitor(desc->block_id().shard_full())) {
auto P = td::PromiseCreator::lambda([](td::Result<td::Ref<ShardState>> R) {
if (R.is_error()) {
auto S = R.move_as_error();
if (S.code() != ErrorCode::timeout && S.code() != ErrorCode::notready) {
VLOG(VALIDATOR_NOTICE) << "failed to get shard state: " << S;
} else {
VLOG(VALIDATOR_DEBUG) << "failed to get shard state: " << S;
}
}
});
wait_block_state_short(desc->block_id(), 0, td::Timestamp::in(60.0), std::move(P));
}
}
@ -554,7 +565,7 @@ void ValidatorManagerImpl::add_cached_block_candidate(ReceivedBlock block) {
if (it != wait_block_data_.end()) {
auto r_block = create_block(cached_block_candidates_[id].clone());
if (r_block.is_ok()) {
td::actor::send_closure(it->second.actor_, &WaitBlockData::got_block_data_from_net, r_block.move_as_ok());
td::actor::send_closure(it->second.actor_, &WaitBlockData::loaded_block_data, r_block.move_as_ok());
}
}
}
@ -672,6 +683,10 @@ void ValidatorManagerImpl::run_ext_query(td::BufferSlice data, td::Promise<td::B
void ValidatorManagerImpl::wait_block_state(BlockHandle handle, td::uint32 priority, td::Timestamp timeout,
td::Promise<td::Ref<ShardState>> promise) {
if (last_masterchain_state_.not_null() && !opts_->need_monitor(handle->id().shard_full(), last_masterchain_state_)) {
return promise.set_error(
td::Status::Error(PSTRING() << "not monitoring shard " << handle->id().shard_full().to_str()));
}
auto it0 = block_state_cache_.find(handle->id());
if (it0 != block_state_cache_.end()) {
it0->second.ttl_ = td::Timestamp::in(30.0);
@ -683,9 +698,11 @@ void ValidatorManagerImpl::wait_block_state(BlockHandle handle, td::uint32 prior
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), handle](td::Result<td::Ref<ShardState>> R) {
td::actor::send_closure(SelfId, &ValidatorManagerImpl::finished_wait_state, handle, std::move(R));
});
auto id = td::actor::create_actor<WaitBlockState>("waitstate", handle, priority, actor_id(this),
td::Timestamp::in(10.0), std::move(P))
.release();
auto id =
td::actor::create_actor<WaitBlockState>("waitstate", handle, priority, actor_id(this),
td::Timestamp::at(timeout.at() + 10.0), std::move(P),
get_block_persistent_state(handle->id()))
.release();
wait_state_[handle->id()].actor_ = id;
it = wait_state_.find(handle->id());
}
@ -717,7 +734,7 @@ void ValidatorManagerImpl::wait_block_data(BlockHandle handle, td::uint32 priori
td::actor::send_closure(SelfId, &ValidatorManagerImpl::finished_wait_data, handle, std::move(R));
});
auto id = td::actor::create_actor<WaitBlockData>("waitdata", handle, priority, actor_id(this),
td::Timestamp::in(10.0), std::move(P))
td::Timestamp::at(timeout.at() + 10.0), false, std::move(P))
.release();
wait_block_data_[handle->id()].actor_ = id;
it = wait_block_data_.find(handle->id());
@ -744,6 +761,10 @@ void ValidatorManagerImpl::wait_block_data_short(BlockIdExt block_id, td::uint32
void ValidatorManagerImpl::wait_block_state_merge(BlockIdExt left_id, BlockIdExt right_id, td::uint32 priority,
td::Timestamp timeout, td::Promise<td::Ref<ShardState>> promise) {
if (last_masterchain_state_.not_null() && !opts_->need_monitor(left_id.shard_full(), last_masterchain_state_)) {
return promise.set_error(
td::Status::Error(PSTRING() << "not monitoring shard " << left_id.shard_full().to_str()));
}
td::actor::create_actor<WaitBlockStateMerge>("merge", left_id, right_id, priority, actor_id(this), timeout,
std::move(promise))
.release();
@ -1128,7 +1149,7 @@ void ValidatorManagerImpl::finished_wait_state(BlockHandle handle, td::Result<td
td::actor::send_closure(SelfId, &ValidatorManagerImpl::finished_wait_state, handle, std::move(R));
});
auto id = td::actor::create_actor<WaitBlockState>("waitstate", handle, X.second, actor_id(this), X.first,
std::move(P))
std::move(P), get_block_persistent_state(handle->id()))
.release();
it->second.actor_ = id;
return;
@ -1158,8 +1179,9 @@ void ValidatorManagerImpl::finished_wait_data(BlockHandle handle, td::Result<td:
td::actor::send_closure(SelfId, &ValidatorManagerImpl::finished_wait_data, handle, std::move(R));
});
auto id =
td::actor::create_actor<WaitBlockData>("waitdata", handle, X.second, actor_id(this), X.first, std::move(P))
.release();
td::actor::create_actor<WaitBlockData>("waitdata", handle, X.second, actor_id(this), X.first, false,
std::move(P))
.release();
it->second.actor_ = id;
return;
}
@ -1617,6 +1639,7 @@ void ValidatorManagerImpl::send_top_shard_block_description(td::Ref<ShardTopBloc
} else {
out_shard_blocks_[ShardTopBlockDescriptionId{desc->block_id().shard_full(), desc->catchain_seqno()}] = desc;
callback_->send_shard_block_info(desc->block_id(), desc->catchain_seqno(), desc->serialize());
add_shard_block_description(desc);
}
}
@ -1629,6 +1652,12 @@ void ValidatorManagerImpl::send_validator_telemetry(PublicKeyHash key,
callback_->send_validator_telemetry(key, std::move(telemetry));
}
void ValidatorManagerImpl::send_download_archive_request(BlockSeqno mc_seqno, ShardIdFull shard_prefix,
std::string tmp_dir, td::Timestamp timeout,
td::Promise<std::string> promise) {
callback_->download_archive(mc_seqno, shard_prefix, std::move(tmp_dir), timeout, std::move(promise));
}
void ValidatorManagerImpl::start_up() {
db_ = create_db_actor(actor_id(this), db_root_, opts_);
actor_stats_ = td::actor::create_actor<td::actor::ActorStats>("actor_stats");
@ -1664,7 +1693,6 @@ void ValidatorManagerImpl::start_up() {
if (fname.substr(fname.size() - 5) != ".pack") {
return;
}
fname = fname.substr(0, fname.size() - 5);
if (fname.substr(0, 8) != "archive.") {
return;
}
@ -1673,13 +1701,18 @@ void ValidatorManagerImpl::start_up() {
while (fname.size() > 1 && fname[0] == '0') {
fname.remove_prefix(1);
}
auto i = fname.find('.');
if (i == td::Slice::npos) {
return;
}
fname = fname.substr(0, i);
auto v = td::to_integer_safe<BlockSeqno>(fname);
if (v.is_error()) {
return;
}
auto pos = v.move_as_ok();
LOG(INFO) << "found archive slice '" << cfname << "' for position " << pos;
to_import_[pos] = std::make_pair(cfname.str(), true);
auto seqno = v.move_as_ok();
LOG(INFO) << "found archive slice '" << cfname << "' for seqno " << seqno;
to_import_[seqno].push_back(cfname.str());
}
});
if (S.is_error()) {
@ -1692,6 +1725,14 @@ void ValidatorManagerImpl::start_up() {
alarm_timestamp().relax(check_waiters_at_);
}
void ValidatorManagerImpl::init_last_masterchain_state(td::Ref<MasterchainState> state) {
if (last_masterchain_state_.not_null()) {
return;
}
last_masterchain_state_ = std::move(state);
update_shard_overlays();
}
void ValidatorManagerImpl::started(ValidatorManagerInitResult R) {
CHECK(R.handle);
CHECK(R.state.not_null());
@ -1730,6 +1771,17 @@ void ValidatorManagerImpl::started(ValidatorManagerInitResult R) {
candidates_buffer_ = td::actor::create_actor<CandidatesBuffer>("candidates-buffer", actor_id(this));
}
init_validator_telemetry();
auto Q = td::PromiseCreator::lambda(
[SelfId = actor_id(this)](td::Result<std::vector<td::Ref<PersistentStateDescription>>> R) {
if (R.is_error()) {
LOG(FATAL) << "db error: " << R.move_as_error();
} else {
td::actor::send_closure(SelfId, &ValidatorManagerImpl::got_persistent_state_descriptions, R.move_as_ok());
}
});
td::actor::send_closure(db_, &Db::get_persistent_state_descriptions, std::move(Q));
update_shard_overlays();
}
void ValidatorManagerImpl::read_gc_list(std::vector<ValidatorSessionId> list) {
@ -1832,61 +1884,35 @@ void ValidatorManagerImpl::download_next_archive() {
}
auto seqno = std::min(last_masterchain_seqno_, shard_client_handle_->id().seqno());
std::vector<std::string> to_import_files;
auto it = to_import_.upper_bound(seqno + 1);
if (it != to_import_.begin()) {
it--;
if (it->second.second) {
it->second.second = false;
downloaded_archive_slice(it->second.first, false);
return;
}
--it;
to_import_files = std::move(it->second);
it->second.clear();
}
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::string> R) {
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::pair<BlockSeqno, BlockSeqno>> R) {
if (R.is_error()) {
LOG(INFO) << "failed to download archive slice: " << R.error();
LOG(INFO) << "failed to download and import archive slice: " << R.error();
delay_action([SelfId]() { td::actor::send_closure(SelfId, &ValidatorManagerImpl::download_next_archive); },
td::Timestamp::in(2.0));
} else {
td::actor::send_closure(SelfId, &ValidatorManagerImpl::downloaded_archive_slice, R.move_as_ok(), true);
td::actor::send_closure(SelfId, &ValidatorManagerImpl::checked_archive_slice, R.ok().first, R.ok().second);
}
});
callback_->download_archive(seqno + 1, db_root_ + "/tmp/", td::Timestamp::in(36000.0), std::move(P));
}
void ValidatorManagerImpl::downloaded_archive_slice(std::string name, bool is_tmp) {
LOG(INFO) << "downloaded archive slice: " << name;
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), name, is_tmp](td::Result<std::vector<BlockSeqno>> R) {
if (is_tmp) {
td::unlink(name).ensure();
}
if (R.is_error()) {
LOG(INFO) << "failed to check downloaded archive slice: " << R.error();
delay_action([SelfId]() { td::actor::send_closure(SelfId, &ValidatorManagerImpl::download_next_archive); },
td::Timestamp::in(2.0));
} else {
td::actor::send_closure(SelfId, &ValidatorManagerImpl::checked_archive_slice, R.move_as_ok());
}
});
auto seqno = std::min(last_masterchain_seqno_, shard_client_handle_->id().seqno());
td::actor::create_actor<ArchiveImporter>("archiveimport", name, last_masterchain_state_, seqno, opts_, actor_id(this),
std::move(P))
td::actor::create_actor<ArchiveImporter>("archiveimport", db_root_, last_masterchain_state_, seqno, opts_,
actor_id(this), std::move(to_import_files), std::move(P))
.release();
}
void ValidatorManagerImpl::checked_archive_slice(std::vector<BlockSeqno> seqno) {
CHECK(seqno.size() == 2);
LOG(INFO) << "checked downloaded archive slice: mc_top_seqno=" << seqno[0] << " shard_top_seqno_=" << seqno[1];
CHECK(seqno[0] <= last_masterchain_seqno_);
CHECK(seqno[1] <= last_masterchain_seqno_);
void ValidatorManagerImpl::checked_archive_slice(BlockSeqno new_last_mc_seqno, BlockSeqno new_shard_client_seqno) {
LOG(INFO) << "checked downloaded archive slice: mc_top_seqno=" << new_last_mc_seqno
<< " shard_top_seqno_=" << new_shard_client_seqno;
CHECK(new_last_mc_seqno <= last_masterchain_seqno_);
CHECK(new_shard_client_seqno <= last_masterchain_seqno_);
BlockIdExt b;
if (seqno[1] < last_masterchain_seqno_) {
CHECK(last_masterchain_state_->get_old_mc_block_id(seqno[1], b));
} else {
b = last_masterchain_block_id_;
}
BlockIdExt shard_client_block_id;
CHECK(last_masterchain_state_->get_old_mc_block_id(new_shard_client_seqno, shard_client_block_id));
auto P = td::PromiseCreator::lambda(
[SelfId = actor_id(this), db = db_.get(), client = shard_client_.get()](td::Result<BlockHandle> R) {
@ -1902,7 +1928,7 @@ void ValidatorManagerImpl::checked_archive_slice(std::vector<BlockSeqno> seqno)
});
td::actor::send_closure(db, &Db::get_block_state, std::move(handle), std::move(P));
});
get_block_handle(b, true, std::move(P));
get_block_handle(shard_client_block_id, true, std::move(P));
}
void ValidatorManagerImpl::finish_prestart_sync() {
@ -1939,6 +1965,7 @@ void ValidatorManagerImpl::new_masterchain_block() {
init_validator_telemetry();
}
update_shard_overlays();
update_shards();
update_shard_blocks();
@ -1952,6 +1979,26 @@ void ValidatorManagerImpl::new_masterchain_block() {
}
}
void ValidatorManagerImpl::update_shard_overlays() {
CHECK(last_masterchain_state_.not_null());
std::set<ShardIdFull> shards_to_monitor;
shards_to_monitor.insert(ShardIdFull{masterchainId});
std::set<WorkchainId> workchains;
for (const auto& shard : last_masterchain_state_->get_shards()) {
workchains.insert(shard->shard().workchain);
if (opts_->need_monitor(shard->shard(),last_masterchain_state_)) {
shards_to_monitor.insert(shard->shard());
}
}
for (const auto &[wc, desc] : last_masterchain_state_->get_workchain_list()) {
if (!workchains.count(wc) && desc->active &&
opts_->need_monitor(ShardIdFull{wc, shardIdAll}, last_masterchain_state_)) {
shards_to_monitor.insert(ShardIdFull{wc, shardIdAll});
}
}
callback_->on_new_masterchain_block(last_masterchain_state_, std::move(shards_to_monitor));
}
void ValidatorManagerImpl::update_shards() {
if ((last_masterchain_state_->rotated_all_shards() || last_masterchain_seqno_ == 0) &&
opts_->get_last_fork_masterchain_seqno() <= last_masterchain_seqno_) {
@ -2268,7 +2315,8 @@ td::actor::ActorOwn<ValidatorGroup> ValidatorManagerImpl::create_validator_group
auto validator_id = get_validator(shard, validator_set);
CHECK(!validator_id.is_zero());
auto G = td::actor::create_actor<ValidatorGroup>(
"validatorgroup", shard, validator_id, session_id, validator_set, key_seqno, opts, keyring_, adnl_, rldp_,
PSTRING() << "valgroup" << shard.to_str(), shard, validator_id, session_id, validator_set, key_seqno, opts,
keyring_, adnl_, rldp_,
overlays_, db_root_, actor_id(this), init_session,
opts_->check_unsafe_resync_allowed(validator_set->get_catchain_seqno()), opts_);
return G;
@ -2444,7 +2492,7 @@ void ValidatorManagerImpl::allow_block_state_gc(BlockIdExt block_id, td::Promise
return;
}
auto shards = gc_masterchain_state_->get_shards();
for (auto shard : shards) {
for (const auto &shard : shards) {
if (shard_intersects(shard->shard(), block_id.shard_full())) {
promise.set_result(block_id.id.seqno < shard->top_block_id().id.seqno);
return;
@ -2682,10 +2730,6 @@ void ValidatorManagerImpl::get_shard_client_state(bool from_db, td::Promise<Bloc
}
}
void ValidatorManagerImpl::subscribe_to_shard(ShardIdFull shard) {
callback_->add_shard(shard);
}
void ValidatorManagerImpl::update_async_serializer_state(AsyncSerializerState state, td::Promise<td::Unit> promise) {
td::actor::send_closure(db_, &Db::update_async_serializer_state, std::move(state), std::move(promise));
}
@ -2698,12 +2742,13 @@ void ValidatorManagerImpl::try_get_static_file(FileHash file_hash, td::Promise<t
td::actor::send_closure(db_, &Db::try_get_static_file, file_hash, std::move(promise));
}
void ValidatorManagerImpl::get_archive_id(BlockSeqno masterchain_seqno, td::Promise<td::uint64> promise) {
void ValidatorManagerImpl::get_archive_id(BlockSeqno masterchain_seqno, ShardIdFull shard_prefix,
td::Promise<td::uint64> promise) {
if (masterchain_seqno > last_masterchain_seqno_) {
promise.set_error(td::Status::Error(ErrorCode::notready, "masterchain seqno too big"));
return;
}
td::actor::send_closure(db_, &Db::get_archive_id, masterchain_seqno, std::move(promise));
td::actor::send_closure(db_, &Db::get_archive_id, masterchain_seqno, shard_prefix, std::move(promise));
}
void ValidatorManagerImpl::get_archive_slice(td::uint64 archive_id, td::uint64 offset, td::uint32 limit,
@ -2715,10 +2760,13 @@ bool ValidatorManagerImpl::is_validator() {
return temp_keys_.size() > 0 || permanent_keys_.size() > 0;
}
bool ValidatorManagerImpl::validating_masterchain() {
return !get_validator(ShardIdFull(masterchainId),
last_masterchain_state_->get_validator_set(ShardIdFull(masterchainId)))
.is_zero();
}
PublicKeyHash ValidatorManagerImpl::get_validator(ShardIdFull shard, td::Ref<ValidatorSet> val_set) {
if (!opts_->need_validate(shard, val_set->get_catchain_seqno())) {
return PublicKeyHash::zero();
}
for (auto &key : temp_keys_) {
if (val_set->is_validator(key.bits256_value())) {
return key;
@ -3224,10 +3272,15 @@ void ValidatorManagerImpl::get_validator_groups_info_for_litequery(
}
void ValidatorManagerImpl::update_options(td::Ref<ValidatorManagerOptions> opts) {
// Currently options can be updated only to change state_serializer_enabled flag and collator_options
if (!shard_client_.empty()) {
td::actor::send_closure(shard_client_, &ShardClient::update_options, opts);
}
if (!serializer_.empty()) {
td::actor::send_closure(serializer_, &AsyncStateSerializer::update_options, opts);
}
if (!queue_size_counter_.empty()) {
td::actor::send_closure(queue_size_counter_, &QueueSizeCounter::update_options, opts);
}
for (auto &group : validator_groups_) {
td::actor::send_closure(group.second.actor, &ValidatorGroup::update_options, opts);
}
@ -3237,6 +3290,53 @@ void ValidatorManagerImpl::update_options(td::Ref<ValidatorManagerOptions> opts)
opts_ = std::move(opts);
}
void ValidatorManagerImpl::add_persistent_state_description(td::Ref<PersistentStateDescription> desc) {
auto now = (UnixTime)td::Clocks::system();
if (desc->end_time <= now) {
return;
}
td::actor::send_closure(db_, &Db::add_persistent_state_description, desc, [](td::Result<td::Unit>) {});
auto it = persistent_state_descriptions_.begin();
while (it != persistent_state_descriptions_.end()) {
const auto &prev_desc = it->second;
if (prev_desc->end_time <= now) {
for (const BlockIdExt &block_id : prev_desc->shard_blocks) {
persistent_state_blocks_.erase(block_id);
}
it = persistent_state_descriptions_.erase(it);
} else {
++it;
}
}
add_persistent_state_description_impl(std::move(desc));
}
void ValidatorManagerImpl::add_persistent_state_description_impl(td::Ref<PersistentStateDescription> desc) {
if (!persistent_state_descriptions_.emplace(desc->masterchain_id.seqno(), desc).second) {
return;
}
LOG(DEBUG) << "Add persistent state description for mc block " << desc->masterchain_id.to_str()
<< " start_time=" << desc->start_time << " end_time=" << desc->end_time;
for (const BlockIdExt &block_id : desc->shard_blocks) {
persistent_state_blocks_[block_id] = desc;
LOG(DEBUG) << "Persistent state description: shard block " << block_id.to_str();
}
}
void ValidatorManagerImpl::got_persistent_state_descriptions(std::vector<td::Ref<PersistentStateDescription>> descs) {
for (auto &desc : descs) {
add_persistent_state_description_impl(std::move(desc));
}
}
td::Ref<PersistentStateDescription> ValidatorManagerImpl::get_block_persistent_state(BlockIdExt block_id) {
auto it = persistent_state_blocks_.find(block_id);
if (it == persistent_state_blocks_.end()) {
return {};
}
return it->second;
}
td::actor::ActorOwn<ValidatorManagerInterface> ValidatorManagerFactory::create(
td::Ref<ValidatorManagerOptions> opts, std::string db_root, td::actor::ActorId<keyring::Keyring> keyring,
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp::Rldp> rldp,