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

New shard overlays

Browse source
This commit is contained in:
SpyCheese 2022-07-29 10:39:02 +03:00
parent 53270a00e6
commit 7ac60bea7d
30 changed files with 411 additions and 271 deletions
Download patch file Download diff file Expand all files Collapse all files

2
create-hardfork/create-hardfork.cpp
View file

@ -236,7 +236,7 @@ class HardforkCreator : public td::actor::Actor {
td::actor::send_closure(id_, &ton::validator::ValidatorManager::sync_complete,
td::PromiseCreator::lambda([](td::Unit) {}));
}
void subscribe_to_shard(ton::ShardIdFull) override {
void update_shard_configuration(td::Ref<ton::validator::MasterchainState> state) override {
}
void send_ihr_message(ton::AccountIdPrefixFull dst, td::BufferSlice data) override {
}

15
overlay/overlay-manager.cpp
View file

@ -92,13 +92,14 @@ void OverlayManager::delete_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdSho
void OverlayManager::create_public_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
std::unique_ptr<Callback> callback, OverlayPrivacyRules rules, td::string scope) {
CHECK(!dht_node_.empty());
CHECK(callback != nullptr);
auto id = overlay_id.compute_short_id();
register_overlay(local_id, id,
Overlay::create(keyring_, adnl_, actor_id(this), dht_node_, local_id, std::move(overlay_id),
std::move(callback), std::move(rules), scope));
}
void OverlayManager::create_public_overlay_no_subscribe(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
void OverlayManager::create_public_overlay_external(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
OverlayPrivacyRules rules, td::string scope) {
CHECK(!dht_node_.empty());
auto id = overlay_id.compute_short_id();
@ -266,12 +267,16 @@ void OverlayManager::get_overlay_random_peers(adnl::AdnlNodeIdShort local_id, Ov
td::uint32 max_peers,
td::Promise<std::vector<adnl::AdnlNodeIdShort>> promise) {
auto it = overlays_.find(local_id);
if (it != overlays_.end()) {
if (it == overlays_.end()) {
promise.set_error(td::Status::Error(PSTRING() << "no such local id " << local_id));
return;
}
auto it2 = it->second.find(overlay_id);
if (it2 != it->second.end()) {
if (it2 == it->second.end()) {
promise.set_error(td::Status::Error(PSTRING() << "no such overlay " << overlay_id));
return;
}
td::actor::send_closure(it2->second, &Overlay::get_overlay_random_peers, max_peers, std::move(promise));
}
}
}
td::actor::ActorOwn<Overlays> Overlays::create(std::string db_root, td::actor::ActorId<keyring::Keyring> keyring,

2
overlay/overlay-manager.h
View file

@ -52,7 +52,7 @@ class OverlayManager : public Overlays {
void create_public_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
std::unique_ptr<Callback> callback, OverlayPrivacyRules rules, td::string scope) override;
void create_public_overlay_no_subscribe(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
void create_public_overlay_external(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
OverlayPrivacyRules rules, td::string scope) override;
void create_private_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
std::vector<adnl::AdnlNodeIdShort> nodes, std::unique_ptr<Callback> callback,

2
overlay/overlay-peers.cpp
View file

@ -142,7 +142,9 @@ void OverlayImpl::receive_random_peers(adnl::AdnlNodeIdShort src, td::BufferSlic
void OverlayImpl::send_random_peers_cont(adnl::AdnlNodeIdShort src, OverlayNode node,
td::Promise<td::BufferSlice> promise) {
std::vector<tl_object_ptr<ton_api::overlay_node>> vec;
if (!is_external()) {
vec.emplace_back(node.tl());
}
for (td::uint32 i = 0; i < nodes_to_send(); i++) {
auto P = get_random_peer();

76
overlay/overlay.cpp
View file

@ -18,6 +18,7 @@
*/
#include "auto/tl/ton_api.h"
#include "td/utils/Random.h"
#include "common/delay.h"
#include "adnl/utils.hpp"
#include "dht/dht.h"
@ -73,7 +74,12 @@ OverlayImpl::OverlayImpl(td::actor::ActorId<keyring::Keyring> keyring, td::actor
, scope_(scope) {
overlay_id_ = id_full_.compute_short_id();
if (is_external()) {
CHECK(public_);
VLOG(OVERLAY_INFO) << this << ": creating public external";
} else {
VLOG(OVERLAY_INFO) << this << ": creating " << (public_ ? "public" : "private");
}
for (auto &node : nodes) {
CHECK(!public_);
@ -86,6 +92,7 @@ OverlayImpl::OverlayImpl(td::actor::ActorId<keyring::Keyring> keyring, td::actor
void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::overlay_getRandomPeers &query,
td::Promise<td::BufferSlice> promise) {
CHECK(!is_external());
if (public_) {
VLOG(OVERLAY_DEBUG) << this << ": received " << query.peers_->nodes_.size() << " nodes from " << src
<< " in getRandomPeers query";
@ -106,6 +113,7 @@ void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::overlay_getR
void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::overlay_getBroadcast &query,
td::Promise<td::BufferSlice> promise) {
CHECK(!is_external());
auto it = broadcasts_.find(query.hash_);
if (it == broadcasts_.end()) {
VLOG(OVERLAY_NOTICE) << this << ": received getBroadcastQuery(" << query.hash_ << ") from " << src
@ -127,16 +135,17 @@ void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::overlay_getB
void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::overlay_getBroadcastList &query,
td::Promise<td::BufferSlice> promise) {
CHECK(!is_external());
VLOG(OVERLAY_WARNING) << this << ": DROPPING getBroadcastList query";
promise.set_error(td::Status::Error(ErrorCode::protoviolation, "dropping get broadcast list query"));
}
/*void OverlayImpl::process_query(adnl::AdnlNodeIdShort src, adnl::AdnlQueryId query_id, ton_api::overlay_customQuery &query) {
callback_->receive_query(src, query_id, id_, std::move(query.data_));
}
*/
void OverlayImpl::receive_query(adnl::AdnlNodeIdShort src, td::BufferSlice data, td::Promise<td::BufferSlice> promise) {
if (is_external()) {
LOG(OVERLAY_WARNING) << "dropping query in external overlay " << overlay_id_;
promise.set_error(td::Status::Error("overlay is external"));
return;
}
if (!public_) {
auto P = peers_.get(src);
if (P == nullptr) {
@ -149,9 +158,6 @@ void OverlayImpl::receive_query(adnl::AdnlNodeIdShort src, td::BufferSlice data,
if (R.is_error()) {
// allow custom query to be here
if (!subscribed()) {
return;
}
callback_->receive_query(src, overlay_id_, std::move(data), std::move(promise));
return;
}
@ -165,27 +171,32 @@ void OverlayImpl::receive_query(adnl::AdnlNodeIdShort src, td::BufferSlice data,
td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
tl_object_ptr<ton_api::overlay_broadcast> bcast) {
CHECK(!is_external());
return BroadcastSimple::create(this, message_from, std::move(bcast));
}
td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
tl_object_ptr<ton_api::overlay_broadcastFec> b) {
CHECK(!is_external());
return OverlayFecBroadcastPart::create(this, message_from, std::move(b));
}
td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
tl_object_ptr<ton_api::overlay_broadcastFecShort> b) {
CHECK(!is_external());
return OverlayFecBroadcastPart::create(this, message_from, std::move(b));
}
td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
tl_object_ptr<ton_api::overlay_broadcastNotFound> bcast) {
CHECK(!is_external());
return td::Status::Error(ErrorCode::protoviolation,
PSTRING() << "received strange message broadcastNotFound from " << message_from);
}
td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
tl_object_ptr<ton_api::overlay_fec_received> msg) {
CHECK(!is_external());
auto it = fec_broadcasts_.find(msg->hash_);
if (it != fec_broadcasts_.end()) {
VLOG(OVERLAY_DEBUG) << this << ": received fec opt-out message from " << message_from << " for broadcast "
@ -200,6 +211,7 @@ td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
tl_object_ptr<ton_api::overlay_fec_completed> msg) {
CHECK(!is_external());
auto it = fec_broadcasts_.find(msg->hash_);
if (it != fec_broadcasts_.end()) {
VLOG(OVERLAY_DEBUG) << this << ": received fec completed message from " << message_from << " for broadcast "
@ -214,12 +226,17 @@ td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
td::Status OverlayImpl::process_broadcast(adnl::AdnlNodeIdShort message_from,
tl_object_ptr<ton_api::overlay_unicast> msg) {
CHECK(!is_external());
VLOG(OVERLAY_DEBUG) << this << ": received unicast from " << message_from;
callback_->receive_message(message_from, overlay_id_, std::move(msg->data_));
return td::Status::OK();
}
void OverlayImpl::receive_message(adnl::AdnlNodeIdShort src, td::BufferSlice data) {
if (is_external()) {
LOG(OVERLAY_WARNING) << "dropping message in external overlay " << overlay_id_;
return;
}
if (!public_) {
if (peers_.get(src) == nullptr) {
VLOG(OVERLAY_WARNING) << this << ": received query in private overlay from unknown source " << src;
@ -228,9 +245,6 @@ void OverlayImpl::receive_message(adnl::AdnlNodeIdShort src, td::BufferSlice dat
}
auto X = fetch_tl_object<ton_api::overlay_Broadcast>(data.clone(), true);
if (X.is_error()) {
if (!subscribed()) {
return;
}
VLOG(OVERLAY_DEBUG) << this << ": received custom message";
callback_->receive_message(src, overlay_id_, std::move(data));
return;
@ -274,7 +288,7 @@ void OverlayImpl::alarm() {
}
if (public_) {
if (peers_.size() > 0 && subscribed()) {
if (peers_.size() > 0) {
auto P = get_random_peer();
if (P) {
send_random_peers(P->get_id(), {});
@ -330,6 +344,10 @@ void OverlayImpl::receive_dht_nodes(td::Result<dht::DhtValue> res, bool dummy) {
VLOG(OVERLAY_NOTICE) << this << ": can not get value from DHT: " << res.move_as_error();
}
if (is_external()) {
return;
}
VLOG(OVERLAY_INFO) << this << ": adding self node to DHT overlay's nodes";
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), oid = print_id()](td::Result<OverlayNode> R) {
if (R.is_error()) {
@ -342,7 +360,7 @@ void OverlayImpl::receive_dht_nodes(td::Result<dht::DhtValue> res, bool dummy) {
}
void OverlayImpl::update_dht_nodes(OverlayNode node) {
if (!public_ || !subscribed()) {
if (!public_) {
return;
}
@ -399,12 +417,30 @@ void OverlayImpl::bcast_gc() {
}
void OverlayImpl::send_message_to_neighbours(td::BufferSlice data) {
if (neighbours_.empty()) {
// TODO: limit retries
delay_action(
[SelfId = actor_id(this), data = std::move(data)]() mutable {
td::actor::send_closure(SelfId, &OverlayImpl::send_message_to_neighbours, std::move(data));
},
td::Timestamp::in(0.5));
return;
}
for (auto &n : neighbours_) {
td::actor::send_closure(manager_, &OverlayManager::send_message, n, local_id_, overlay_id_, data.clone());
}
}
void OverlayImpl::send_broadcast(PublicKeyHash send_as, td::uint32 flags, td::BufferSlice data) {
if (neighbours_.empty()) {
// TODO: limit retries
delay_action(
[SelfId = actor_id(this), send_as, flags, data = std::move(data)]() mutable {
td::actor::send_closure(SelfId, &OverlayImpl::send_broadcast, send_as, flags, std::move(data));
},
td::Timestamp::in(0.5));
return;
}
auto S = BroadcastSimple::create_new(actor_id(this), keyring_, send_as, std::move(data), flags);
if (S.is_error()) {
LOG(WARNING) << "failed to send broadcast: " << S;
@ -412,6 +448,15 @@ void OverlayImpl::send_broadcast(PublicKeyHash send_as, td::uint32 flags, td::Bu
}
void OverlayImpl::send_broadcast_fec(PublicKeyHash send_as, td::uint32 flags, td::BufferSlice data) {
if (neighbours_.empty()) {
// TODO: limit retries
delay_action(
[SelfId = actor_id(this), send_as, flags, data = std::move(data)]() mutable {
td::actor::send_closure(SelfId, &OverlayImpl::send_broadcast_fec, send_as, flags, std::move(data));
},
td::Timestamp::in(0.5));
return;
}
OverlayOutboundFecBroadcast::create(std::move(data), flags, actor_id(this), send_as);
}
@ -503,7 +548,7 @@ void OverlayImpl::send_new_fec_broadcast_part(PublicKeyHash local_id, Overlay::B
}
void OverlayImpl::deliver_broadcast(PublicKeyHash source, td::BufferSlice data) {
if (!subscribed()) {
if (is_external()) {
return;
}
callback_->receive_broadcast(source, overlay_id_, std::move(data));
@ -578,7 +623,8 @@ void OverlayImpl::set_privacy_rules(OverlayPrivacyRules rules) {
}
void OverlayImpl::check_broadcast(PublicKeyHash src, td::BufferSlice data, td::Promise<td::Unit> promise) {
if (!subscribed()) {
if (is_external()) {
promise.set_result(td::Unit());
return;
}
callback_->check_broadcast(src, overlay_id_, std::move(data), std::move(promise));

8
overlay/overlay.hpp
View file

@ -251,15 +251,13 @@ class OverlayImpl : public Overlay {
}
private:
bool subscribed() const {
return (bool)callback_;
bool is_external() const {
return callback_ == nullptr;
}
template <class T>
void process_query(adnl::AdnlNodeIdShort src, T &query, td::Promise<td::BufferSlice> promise) {
if (!subscribed()) {
return;
}
CHECK(!is_external());
callback_->receive_query(src, overlay_id_, serialize_tl_object(&query, true), std::move(promise));
}

2
overlay/overlays.h
View file

@ -194,7 +194,7 @@ class Overlays : public td::actor::Actor {
virtual void create_public_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
std::unique_ptr<Callback> callback, OverlayPrivacyRules rules, td::string scope) = 0;
virtual void create_public_overlay_no_subscribe(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
virtual void create_public_overlay_external(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
OverlayPrivacyRules rules, td::string scope) = 0;
virtual void create_private_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
std::vector<adnl::AdnlNodeIdShort> nodes, std::unique_ptr<Callback> callback,

2
test/test-ton-collator.cpp
View file

@ -323,7 +323,7 @@ class TestNode : public td::actor::Actor {
td::actor::send_closure(id_, &ton::validator::ValidatorManager::sync_complete,
td::PromiseCreator::lambda([](td::Unit) {}));
}
void subscribe_to_shard(ton::ShardIdFull) override {
void update_shard_configuration(td::Ref<ton::validator::MasterchainState> state) override {
}
void send_ihr_message(ton::AccountIdPrefixFull dst, td::BufferSlice data) override {
}

2
ton/ton-types.h
View file

@ -57,6 +57,8 @@ constexpr unsigned min_split_merge_interval = 30; // split/merge interval must
constexpr unsigned max_split_merge_delay =
1000; // end of split/merge interval must be at most 1000 seconds in the future
constexpr int max_shard_pfx_len = 60;
enum GlobalCapabilities {
capIhrEnabled = 1,
capCreateStatsEnabled = 2,

59
validator-engine/validator-engine.cpp
View file

@ -1324,18 +1324,6 @@ td::Status ValidatorEngine::load_global_config() {
}
validator_options_ = ton::validator::ValidatorManagerOptions::create(zero_state, init_block);
validator_options_.write().set_shard_check_function(
[masterchain_only = masterchain_only_](ton::ShardIdFull shard, ton::CatchainSeqno cc_seqno,
ton::validator::ValidatorManagerOptions::ShardCheckMode mode) -> bool {
if (mode == ton::validator::ValidatorManagerOptions::ShardCheckMode::m_monitor) {
return shard.is_masterchain() || !masterchain_only;
}
CHECK(mode == ton::validator::ValidatorManagerOptions::ShardCheckMode::m_validate);
return true;
/*ton::ShardIdFull p{ton::basechainId, ((cc_seqno * 1ull % 4) << 62) + 1};
auto s = ton::shard_prefix(p, 2);
return shard.is_masterchain() || ton::shard_intersects(shard, s);*/
});
if (state_ttl_ != 0) {
validator_options_.write().set_state_ttl(state_ttl_);
}
@ -1389,6 +1377,39 @@ td::Status ValidatorEngine::load_global_config() {
return td::Status::OK();
}
void ValidatorEngine::init_validator_options() {
if (!masterchain_only_) {
validator_options_.write().set_shard_check_function(
[](ton::ShardIdFull shard, ton::CatchainSeqno cc_seqno,
ton::validator::ValidatorManagerOptions::ShardCheckMode mode) -> bool {
if (mode == ton::validator::ValidatorManagerOptions::ShardCheckMode::m_monitor) {
return true;
}
CHECK(mode == ton::validator::ValidatorManagerOptions::ShardCheckMode::m_validate);
return true;
});
} else {
std::vector<ton::ShardIdFull> shards = {ton::ShardIdFull(ton::masterchainId)};
for (const auto& c : config_.collators) {
shards.push_back(c.shard);
}
validator_options_.write().set_shard_check_function(
[shards = std::move(shards)](ton::ShardIdFull shard, ton::CatchainSeqno cc_seqno,
ton::validator::ValidatorManagerOptions::ShardCheckMode mode) -> bool {
if (mode == ton::validator::ValidatorManagerOptions::ShardCheckMode::m_monitor) {
for (auto s : shards) {
if (shard_is_ancestor(shard, s)) {
return true;
}
}
return false;
}
CHECK(mode == ton::validator::ValidatorManagerOptions::ShardCheckMode::m_validate);
return true;
});
}
}
void ValidatorEngine::load_empty_local_config(td::Promise<td::Unit> promise) {
auto ret_promise = td::PromiseCreator::lambda(
[SelfId = actor_id(this), promise = std::move(promise)](td::Result<td::Unit> R) mutable {
@ -1671,6 +1692,7 @@ void ValidatorEngine::got_key(ton::PublicKey key) {
}
void ValidatorEngine::start() {
init_validator_options();
read_config_ = true;
start_adnl();
}
@ -1825,19 +1847,22 @@ void ValidatorEngine::start_full_node() {
};
full_node_client_ = ton::adnl::AdnlExtMultiClient::create(std::move(vec), std::make_unique<Cb>());
}
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::Unit> R) {
R.ensure();
td::actor::send_closure(SelfId, &ValidatorEngine::started_full_node);
});
full_node_ = ton::validator::fullnode::FullNode::create(
short_id, ton::adnl::AdnlNodeIdShort{config_.full_node}, validator_options_->zero_block_id().file_hash,
keyring_.get(), adnl_.get(), rldp_.get(),
validator_options_, keyring_.get(), adnl_.get(), rldp_.get(),
default_dht_node_.is_zero() ? td::actor::ActorId<ton::dht::Dht>{} : dht_nodes_[default_dht_node_].get(),
overlay_manager_.get(), validator_manager_.get(), full_node_client_.get(), db_root_);
}
overlay_manager_.get(), validator_manager_.get(), full_node_client_.get(), db_root_, std::move(P));
for (auto &v : config_.validators) {
td::actor::send_closure(full_node_, &ton::validator::fullnode::FullNode::add_permanent_key, v.first,
[](td::Unit) {});
}
} else {
started_full_node();
}
}
void ValidatorEngine::started_full_node() {

1
validator-engine/validator-engine.hpp
View file

@ -276,6 +276,7 @@ class ValidatorEngine : public td::actor::Actor {
void load_empty_local_config(td::Promise<td::Unit> promise);
void load_local_config(td::Promise<td::Unit> promise);
void load_config(td::Promise<td::Unit> promise);
void init_validator_options();
void start();

4
validator/fabric.h
View file

@ -52,7 +52,7 @@ void run_check_external_message(td::BufferSlice data, td::actor::ActorId<Validat
void run_accept_block_query(BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::Ref<BlockSignatureSet> signatures,
td::Ref<BlockSignatureSet> approve_signatures, bool send_broadcast,
td::Ref<BlockSignatureSet> approve_signatures, bool send_broadcast, bool apply,
td::actor::ActorId<ValidatorManager> manager, td::Promise<td::Unit> promise);
void run_fake_accept_block_query(BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::actor::ActorId<ValidatorManager> manager,
@ -61,7 +61,7 @@ void run_hardfork_accept_block_query(BlockIdExt id, td::Ref<BlockData> data,
td::actor::ActorId<ValidatorManager> manager, td::Promise<td::Unit> promise);
void run_broadcast_only_accept_block_query(BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::Ref<BlockSignatureSet> signatures,
td::Ref<BlockSignatureSet> approve_signatures,
td::Ref<BlockSignatureSet> approve_signatures, bool send_block_broadcast,
td::actor::ActorId<ValidatorManager> manager, td::Promise<td::Unit> promise);
void run_apply_block_query(BlockIdExt id, td::Ref<BlockData> block, BlockIdExt masterchain_block_id,
td::actor::ActorId<ValidatorManager> manager, td::Timestamp timeout,

44
validator/full-node-shard.cpp
View file

@ -69,7 +69,7 @@ void Neighbour::update_roundtrip(double t) {
}
void FullNodeShardImpl::create_overlay() {
if (subscribed_) {
if (active_) {
class Callback : public overlay::Overlays::Callback {
public:
void receive_message(adnl::AdnlNodeIdShort src, overlay::OverlayIdShort overlay_id,
@ -99,7 +99,7 @@ void FullNodeShardImpl::create_overlay() {
PSTRING() << "{ \"type\": \"shard\", \"shard_id\": " << get_shard()
<< ", \"workchain_id\": " << get_workchain() << " }");
} else {
td::actor::send_closure(overlays_, &overlay::Overlays::create_public_overlay_no_subscribe, adnl_id_,
td::actor::send_closure(overlays_, &overlay::Overlays::create_public_overlay_external, adnl_id_,
overlay_id_full_.clone(), rules_,
PSTRING() << "{ \"type\": \"shard\", \"shard_id\": " << get_shard()
<< ", \"workchain_id\": " << get_workchain() << " }");
@ -129,12 +129,12 @@ void FullNodeShardImpl::update_adnl_id(adnl::AdnlNodeIdShort adnl_id, td::Promis
create_overlay();
}
void FullNodeShardImpl::subscribe_to_shard() {
if (subscribed_ || !client_.empty()) {
void FullNodeShardImpl::set_active(bool active) {
if (active_ == active || shard_.is_masterchain()) {
return;
}
td::actor::send_closure(overlays_, &ton::overlay::Overlays::delete_overlay, adnl_id_, overlay_id_);
subscribed_ = true;
active_ = active;
create_overlay();
}
@ -576,6 +576,10 @@ void FullNodeShardImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::tonNod
void FullNodeShardImpl::receive_query(adnl::AdnlNodeIdShort src, td::BufferSlice query,
td::Promise<td::BufferSlice> promise) {
if (!active_) {
promise.set_error(td::Status::Error("shard is inactive"));
return;
}
auto B = fetch_tl_object<ton_api::Function>(std::move(query), true);
if (B.is_error()) {
promise.set_error(td::Status::Error(ErrorCode::protoviolation, "cannot parse tonnode query"));
@ -596,28 +600,23 @@ void FullNodeShardImpl::process_broadcast(PublicKeyHash src, ton_api::tonNode_ex
void FullNodeShardImpl::process_broadcast(PublicKeyHash src, ton_api::tonNode_newShardBlockBroadcast &query) {
auto block_id = create_block_id(query.block_->block_);
if (block_id.shard_full() != shard_) {
LOG(WARNING) << "dropping newShardBlockBroadcast: expected shard " << shard_.to_str() << ", got shard "
<< block_id.shard_full().to_str();
return;
}
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::new_shard_block,
block_id, query.block_->cc_seqno_,
std::move(query.block_->data_));
}
void FullNodeShardImpl::process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcast &query) {
BlockIdExt block_id = create_block_id(query.id_);
if (block_id.shard_full() != shard_) {
LOG(FULL_NODE_WARNING) << "dropping block broadcast: shard mismatch. overlay=" << shard_.to_str()
<< " block=" << block_id.to_str();
return;
}
std::vector<BlockSignature> signatures;
for (auto &sig : query.signatures_) {
signatures.emplace_back(BlockSignature{sig->who_, std::move(sig->signature_)});
}
BlockIdExt block_id = create_block_id(query.id_);
if (block_id.shard_full() != shard_) {
LOG(WARNING) << "dropping blockBroadcast: expected shard " << shard_.to_str() << ", got shard "
<< block_id.shard_full().to_str();
return;
}
BlockBroadcast B{block_id,
std::move(signatures),
static_cast<UnixTime>(query.catchain_seqno_),
@ -639,6 +638,9 @@ void FullNodeShardImpl::process_broadcast(PublicKeyHash src, ton_api::tonNode_bl
}
void FullNodeShardImpl::receive_broadcast(PublicKeyHash src, td::BufferSlice broadcast) {
if (!active_) {
return;
}
auto B = fetch_tl_object<ton_api::tonNode_Broadcast>(std::move(broadcast), true);
if (B.is_error()) {
return;
@ -1087,7 +1089,7 @@ FullNodeShardImpl::FullNodeShardImpl(ShardIdFull shard, PublicKeyHash local_id,
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp::Rldp> rldp,
td::actor::ActorId<overlay::Overlays> overlays,
td::actor::ActorId<ValidatorManagerInterface> validator_manager,
td::actor::ActorId<adnl::AdnlExtClient> client, bool subscribe)
td::actor::ActorId<adnl::AdnlExtClient> client, bool active)
: shard_(shard)
, local_id_(local_id)
, adnl_id_(adnl_id)
@ -1098,7 +1100,7 @@ FullNodeShardImpl::FullNodeShardImpl(ShardIdFull shard, PublicKeyHash local_id,
, overlays_(overlays)
, validator_manager_(validator_manager)
, client_(client)
, subscribed_(subscribe) {
, active_(shard.is_masterchain() || active) {
}
td::actor::ActorOwn<FullNodeShard> FullNodeShard::create(
@ -1106,9 +1108,9 @@ td::actor::ActorOwn<FullNodeShard> FullNodeShard::create(
td::actor::ActorId<keyring::Keyring> keyring, td::actor::ActorId<adnl::Adnl> adnl,
td::actor::ActorId<rldp::Rldp> rldp, td::actor::ActorId<overlay::Overlays> overlays,
td::actor::ActorId<ValidatorManagerInterface> validator_manager, td::actor::ActorId<adnl::AdnlExtClient> client,
bool subscribe) {
bool active) {
return td::actor::create_actor<FullNodeShardImpl>("tonnode", shard, local_id, adnl_id, zero_state_file_hash, keyring,
adnl, rldp, overlays, validator_manager, client, subscribe);
adnl, rldp, overlays, validator_manager, client, active);
}
} // namespace fullnode

4
validator/full-node-shard.h
View file

@ -36,7 +36,7 @@ class FullNodeShard : public td::actor::Actor {
virtual ShardIdFull get_shard_full() const = 0;
virtual void update_adnl_id(adnl::AdnlNodeIdShort adnl_id, td::Promise<td::Unit> promise) = 0;
virtual void subscribe_to_shard() = 0;
virtual void set_active(bool active) = 0;
virtual void send_ihr_message(td::BufferSlice data) = 0;
virtual void send_external_message(td::BufferSlice data) = 0;
@ -72,7 +72,7 @@ class FullNodeShard : public td::actor::Actor {
td::actor::ActorId<keyring::Keyring> keyring, td::actor::ActorId<adnl::Adnl> adnl,
td::actor::ActorId<rldp::Rldp> rldp, td::actor::ActorId<overlay::Overlays> overlays,
td::actor::ActorId<ValidatorManagerInterface> validator_manager, td::actor::ActorId<adnl::AdnlExtClient> client,
bool subscribe);
bool active = true);
};
} // namespace fullnode

6
validator/full-node-shard.hpp
View file

@ -80,7 +80,7 @@ class FullNodeShardImpl : public FullNodeShard {
void create_overlay();
void update_adnl_id(adnl::AdnlNodeIdShort adnl_id, td::Promise<td::Unit> promise) override;
virtual void subscribe_to_shard() override;
void set_active(bool active) override;
//td::Result<Block> fetch_block(td::BufferSlice data);
void prevalidate_block(BlockIdExt block_id, td::BufferSlice data, td::BufferSlice proof,
@ -204,7 +204,7 @@ class FullNodeShardImpl : public FullNodeShard {
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp::Rldp> rldp,
td::actor::ActorId<overlay::Overlays> overlays,
td::actor::ActorId<ValidatorManagerInterface> validator_manager,
td::actor::ActorId<adnl::AdnlExtClient> client, bool subscribe);
td::actor::ActorId<adnl::AdnlExtClient> client, bool active = true);
private:
bool use_new_download() const {
@ -242,7 +242,7 @@ class FullNodeShardImpl : public FullNodeShard {
td::Timestamp ping_neighbours_at_;
adnl::AdnlNodeIdShort last_pinged_neighbour_ = adnl::AdnlNodeIdShort::zero();
bool subscribed_ = false;
bool active_ = false;
};
} // namespace fullnode

229
validator/full-node.cpp
View file

@ -20,6 +20,7 @@
#include "ton/ton-shard.h"
#include "ton/ton-io.hpp"
#include "td/actor/MultiPromise.h"
#include "ton/ton-types.h"
namespace ton {
@ -27,6 +28,8 @@ namespace validator {
namespace fullnode {
static const double INACTIVE_SHARD_TTL = 120.0;
void FullNodeImpl::add_permanent_key(PublicKeyHash key, td::Promise<td::Unit> promise) {
if (local_keys_.count(key)) {
promise.set_value(td::Unit());
@ -47,7 +50,9 @@ void FullNodeImpl::add_permanent_key(PublicKeyHash key, td::Promise<td::Unit> pr
}
for (auto &shard : shards_) {
td::actor::send_closure(shard.second, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
if (!shard.second.actor.empty()) {
td::actor::send_closure(shard.second.actor, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
}
}
promise.set_value(td::Unit());
}
@ -71,7 +76,9 @@ void FullNodeImpl::del_permanent_key(PublicKeyHash key, td::Promise<td::Unit> pr
}
for (auto &shard : shards_) {
td::actor::send_closure(shard.second, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
if (!shard.second.actor.empty()) {
td::actor::send_closure(shard.second.actor, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
}
}
promise.set_value(td::Unit());
}
@ -80,21 +87,24 @@ void FullNodeImpl::sign_shard_overlay_certificate(ShardIdFull shard_id, PublicKe
td::uint32 expiry_at, td::uint32 max_size,
td::Promise<td::BufferSlice> promise) {
auto it = shards_.find(shard_id);
if(it == shards_.end()) {
if(it == shards_.end() || it->second.actor.empty()) {
promise.set_error(td::Status::Error(ErrorCode::error, "shard not found"));
return;
}
td::actor::send_closure(it->second, &FullNodeShard::sign_overlay_certificate, signed_key, expiry_at, max_size, std::move(promise));
td::actor::send_closure(it->second.actor, &FullNodeShard::sign_overlay_certificate, signed_key, expiry_at, max_size,
std::move(promise));
}
void FullNodeImpl::import_shard_overlay_certificate(ShardIdFull shard_id, PublicKeyHash signed_key,
std::shared_ptr<ton::overlay::Certificate> cert,
td::Promise<td::Unit> promise) {
auto it = shards_.find(shard_id);
if(it == shards_.end()) {
if(it == shards_.end() || it->second.actor.empty()) {
promise.set_error(td::Status::Error(ErrorCode::error, "shard not found"));
return;
}
td::actor::send_closure(it->second, &FullNodeShard::import_overlay_certificate, signed_key, cert, std::move(promise));
td::actor::send_closure(it->second.actor, &FullNodeShard::import_overlay_certificate, signed_key, cert,
std::move(promise));
}
void FullNodeImpl::update_adnl_id(adnl::AdnlNodeIdShort adnl_id, td::Promise<td::Unit> promise) {
@ -105,7 +115,9 @@ void FullNodeImpl::update_adnl_id(adnl::AdnlNodeIdShort adnl_id, td::Promise<td:
ig.add_promise(std::move(promise));
for (auto &s : shards_) {
td::actor::send_closure(s.second, &FullNodeShard::update_adnl_id, adnl_id, ig.get_promise());
if (!s.second.actor.empty()) {
td::actor::send_closure(s.second.actor, &FullNodeShard::update_adnl_id, adnl_id, ig.get_promise());
}
}
local_id_ = adnl_id_.pubkey_hash();
}
@ -116,34 +128,116 @@ void FullNodeImpl::initial_read_complete(BlockHandle top_handle) {
td::actor::send_closure(SelfId, &FullNodeImpl::sync_completed);
});
auto it = shards_.find(ShardIdFull{masterchainId});
CHECK(it != shards_.end());
td::actor::send_closure(it->second, &FullNodeShard::set_handle, top_handle, std::move(P));
CHECK(it != shards_.end() && !it->second.actor.empty());
td::actor::send_closure(it->second.actor, &FullNodeShard::set_handle, top_handle, std::move(P));
}
void FullNodeImpl::add_shard(ShardIdFull shard, bool subscribe) {
while (true) {
auto it = shards_.find(shard);
if (it == shards_.end()) {
shards_.emplace(shard, FullNodeShard::create(shard, local_id_, adnl_id_, zero_state_file_hash_, keyring_, adnl_,
rldp_, overlays_, validator_manager_, client_, subscribe));
if (all_validators_.size() > 0) {
td::actor::send_closure(shards_[shard], &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
}
} else if (subscribe) {
td::actor::send_closure(it->second, &FullNodeShard::subscribe_to_shard);
} else {
break;
}
if (shard.shard == shardIdAll) {
break;
}
void FullNodeImpl::update_shard_configuration(td::Ref<MasterchainState> state) {
std::map<ShardIdFull, BlockIdExt> new_shards;
std::set<ShardIdFull> new_active;
new_shards[ShardIdFull(masterchainId)] = state->get_block_id();
new_active.insert(ShardIdFull(masterchainId));
std::set<WorkchainId> workchains;
auto cur_time = state->get_unix_time();
auto set_active = [&](ShardIdFull shard) {
while (new_active.insert(shard).second && shard.pfx_len() > 0) {
shard = shard_parent(shard);
}
};
for (auto &info : state->get_shards()) {
auto shard = info->shard();
workchains.insert(shard.workchain);
new_shards[shard] = info->top_block_id();
bool will_split = shard.pfx_len() < max_shard_pfx_len && ((info->fsm_state() == McShardHash::FsmState::fsm_split &&
info->fsm_utime() < cur_time + 60) || info->before_split());
bool will_merge = shard.pfx_len() > 0 && ((info->fsm_state() == McShardHash::FsmState::fsm_merge &&
info->fsm_utime() < cur_time + 60) || info->before_merge());
if (opts_->need_monitor(shard)) {
set_active(shard);
}
if (will_merge && opts_->need_monitor(shard_parent(shard))) {
set_active(shard);
set_active(shard_sibling(shard));
}
for (int id = 0; id < 2; ++id) {
if (will_split && opts_->need_monitor(shard_child(shard, id))) {
set_active(shard_child(shard, id));
}
}
}
for (const auto &wpair : state->get_workchain_list()) {
ton::WorkchainId wc = wpair.first;
const block::WorkchainInfo *winfo = wpair.second.get();
if (workchains.count(wc) == 0 && winfo->active && winfo->enabled_since <= cur_time) {
auto shard = ShardIdFull(wc);
new_shards[shard] = BlockIdExt(wc, shard.shard, 0, winfo->zerostate_root_hash, winfo->zerostate_file_hash);
if (opts_->need_monitor(shard)) {
set_active(shard);
}
}
}
auto info_set_active = [&](ShardIdFull shard, ShardInfo& info, bool active) {
if (info.active == active) {
return;
}
if (info.actor.empty()) {
add_shard_actor(shard, active);
return;
}
info.active = active;
td::actor::send_closure(info.actor, &FullNodeShard::set_active, active);
info.delete_at = active ? td::Timestamp::never() : td::Timestamp::in(INACTIVE_SHARD_TTL);
};
for (auto shard : new_shards) {
auto &info = shards_[shard.first];
info.exists = true;
if (!info.active && new_active.count(shard.first)) {
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::wait_block_state_short, shard.second, 0,
td::Timestamp::in(60.0), [](td::Result<td::Ref<ShardState>>){});
}
}
for (auto& p : shards_) {
ShardIdFull shard = p.first;
ShardInfo &info = p.second;
info.exists = new_shards.count(shard);
info_set_active(shard, info, new_active.count(shard));
}
for (ShardIdFull shard : new_active) {
info_set_active(shard, shards_[shard], true);
}
auto it = shards_.begin();
while (it != shards_.end()) {
if (!it->second.active && it->second.delete_at && it->second.delete_at.is_in_past()) {
it->second.actor.reset();
it->second.delete_at = td::Timestamp::never();
}
if (!it->second.exists && it->second.actor.empty()) {
it = shards_.erase(it);
} else {
++it;
}
}
}
void FullNodeImpl::del_shard(ShardIdFull shard) {
LOG(FATAL) << "deleting shards not implemented: shard=" << shard;
shards_.erase(shard);
void FullNodeImpl::add_shard_actor(ShardIdFull shard, bool active) {
ShardInfo &info = shards_[shard];
if (!info.actor.empty()) {
return;
}
info.actor = FullNodeShard::create(shard, local_id_, adnl_id_, zero_state_file_hash_, keyring_, adnl_, rldp_,
overlays_, validator_manager_, client_, active);
info.active = active;
info.delete_at = active ? td::Timestamp::never() : td::Timestamp::in(INACTIVE_SHARD_TTL);
if (all_validators_.size() > 0) {
td::actor::send_closure(info.actor, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
}
}
void FullNodeImpl::sync_completed() {
@ -169,7 +263,7 @@ void FullNodeImpl::send_ext_message(AccountIdPrefixFull dst, td::BufferSlice dat
}
void FullNodeImpl::send_shard_block_info(BlockIdExt block_id, CatchainSeqno cc_seqno, td::BufferSlice data) {
auto shard = get_shard(block_id.shard_full());
auto shard = get_shard(ShardIdFull{masterchainId});
if (shard.empty()) {
VLOG(FULL_NODE_WARNING) << "dropping OUT shard block info message to unknown shard";
return;
@ -178,7 +272,7 @@ void FullNodeImpl::send_shard_block_info(BlockIdExt block_id, CatchainSeqno cc_s
}
void FullNodeImpl::send_broadcast(BlockBroadcast broadcast) {
auto shard = get_shard(broadcast.block_id.shard_full());
auto shard = get_shard(broadcast.block_id.shard_full(), true);
if (shard.empty()) {
VLOG(FULL_NODE_WARNING) << "dropping OUT broadcast to unknown shard";
return;
@ -262,19 +356,38 @@ void FullNodeImpl::download_archive(BlockSeqno masterchain_seqno, std::string tm
std::move(promise));
}
td::actor::ActorId<FullNodeShard> FullNodeImpl::get_shard(ShardIdFull shard) {
add_shard(ShardIdFull{shard.workchain, shardIdAll});
while (shards_.count(shard) == 0) {
if (shard.shard == shardIdAll) {
return td::actor::ActorId<FullNodeShard>{};
td::actor::ActorId<FullNodeShard> FullNodeImpl::get_shard(ShardIdFull shard, bool exact) {
if (!exact) {
ShardIdFull s = shard;
while (true) {
auto it = shards_.find(s);
if (it != shards_.end() && it->second.exists) {
if (it->second.actor.empty()) {
add_shard_actor(s, false);
}
shard = shard_parent(shard);
if (!it->second.active) {
it->second.delete_at = td::Timestamp::in(INACTIVE_SHARD_TTL);
}
return shards_[shard].get();
return it->second.actor.get();
}
if (s.pfx_len() == 0) {
break;
}
s = shard_parent(s);
}
}
auto &info = shards_[shard];
if (info.actor.empty()) {
add_shard_actor(shard, false);
}
if (!info.active) {
info.delete_at = td::Timestamp::in(INACTIVE_SHARD_TTL);
}
return info.actor.get();
}
td::actor::ActorId<FullNodeShard> FullNodeImpl::get_shard(AccountIdPrefixFull dst) {
return get_shard(shard_prefix(dst, 60));
return get_shard(shard_prefix(dst, max_shard_pfx_len));
}
void FullNodeImpl::got_key_block_proof(td::Ref<ProofLink> proof) {
@ -307,7 +420,7 @@ void FullNodeImpl::got_key_block_proof(td::Ref<ProofLink> proof) {
CHECK(all_validators_.size() > 0);
for (auto &shard : shards_) {
td::actor::send_closure(shard.second, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
td::actor::send_closure(shard.second.actor, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
}
}
@ -339,7 +452,9 @@ void FullNodeImpl::got_zero_block_state(td::Ref<ShardState> state) {
CHECK(all_validators_.size() > 0);
for (auto &shard : shards_) {
td::actor::send_closure(shard.second, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
if (!shard.second.actor.empty()) {
td::actor::send_closure(shard.second.actor, &FullNodeShard::update_validators, all_validators_, sign_cert_by_);
}
}
}
@ -384,8 +499,8 @@ void FullNodeImpl::start_up() {
void initial_read_complete(BlockHandle handle) override {
td::actor::send_closure(id_, &FullNodeImpl::initial_read_complete, handle);
}
void subscribe_to_shard(ShardIdFull shard) override {
td::actor::send_closure(id_, &FullNodeImpl::add_shard, shard, true);
void update_shard_configuration(td::Ref<MasterchainState> state) override {
td::actor::send_closure(id_, &FullNodeImpl::update_shard_configuration, std::move(state));
}
void send_ihr_message(AccountIdPrefixFull dst, td::BufferSlice data) override {
td::actor::send_closure(id_, &FullNodeImpl::send_ihr_message, dst, std::move(data));
@ -443,20 +558,21 @@ void FullNodeImpl::start_up() {
td::actor::ActorId<FullNodeImpl> id_;
};
auto P = td::PromiseCreator::lambda([](td::Unit R) {});
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::install_callback,
std::make_unique<Callback>(actor_id(this)), std::move(P));
std::make_unique<Callback>(actor_id(this)), std::move(started_promise_));
}
FullNodeImpl::FullNodeImpl(PublicKeyHash local_id, adnl::AdnlNodeIdShort adnl_id, FileHash zero_state_file_hash,
td::actor::ActorId<keyring::Keyring> keyring, td::actor::ActorId<adnl::Adnl> adnl,
td::actor::ActorId<rldp::Rldp> rldp, td::actor::ActorId<dht::Dht> dht,
td::actor::ActorId<overlay::Overlays> overlays,
td::Ref<ValidatorManagerOptions> opts, td::actor::ActorId<keyring::Keyring> keyring,
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp::Rldp> rldp,
td::actor::ActorId<dht::Dht> dht, td::actor::ActorId<overlay::Overlays> overlays,
td::actor::ActorId<ValidatorManagerInterface> validator_manager,
td::actor::ActorId<adnl::AdnlExtClient> client, std::string db_root)
td::actor::ActorId<adnl::AdnlExtClient> client, std::string db_root,
td::Promise<td::Unit> started_promise)
: local_id_(local_id)
, adnl_id_(adnl_id)
, zero_state_file_hash_(zero_state_file_hash)
, opts_(opts)
, keyring_(keyring)
, adnl_(adnl)
, rldp_(rldp)
@ -464,20 +580,23 @@ FullNodeImpl::FullNodeImpl(PublicKeyHash local_id, adnl::AdnlNodeIdShort adnl_id
, overlays_(overlays)
, validator_manager_(validator_manager)
, client_(client)
, db_root_(db_root) {
add_shard(ShardIdFull{masterchainId}, true);
, db_root_(db_root)
, started_promise_(std::move(started_promise)) {
add_shard_actor(ShardIdFull{masterchainId}, true);
}
td::actor::ActorOwn<FullNode> FullNode::create(ton::PublicKeyHash local_id, adnl::AdnlNodeIdShort adnl_id,
FileHash zero_state_file_hash,
FileHash zero_state_file_hash, td::Ref<ValidatorManagerOptions> opts,
td::actor::ActorId<keyring::Keyring> keyring,
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp::Rldp> rldp,
td::actor::ActorId<dht::Dht> dht,
td::actor::ActorId<overlay::Overlays> overlays,
td::actor::ActorId<ValidatorManagerInterface> validator_manager,
td::actor::ActorId<adnl::AdnlExtClient> client, std::string db_root) {
return td::actor::create_actor<FullNodeImpl>("fullnode", local_id, adnl_id, zero_state_file_hash, keyring, adnl, rldp,
dht, overlays, validator_manager, client, db_root);
td::actor::ActorId<adnl::AdnlExtClient> client, std::string db_root,
td::Promise<td::Unit> started_promise) {
return td::actor::create_actor<FullNodeImpl>("fullnode", local_id, adnl_id, zero_state_file_hash, opts, keyring, adnl,
rldp, dht, overlays, validator_manager, client, db_root,
std::move(started_promise));
}
} // namespace fullnode

5
validator/full-node.h
View file

@ -73,13 +73,14 @@ class FullNode : public td::actor::Actor {
}
static td::actor::ActorOwn<FullNode> create(ton::PublicKeyHash local_id, adnl::AdnlNodeIdShort adnl_id,
FileHash zero_state_file_hash,
FileHash zero_state_file_hash, td::Ref<ValidatorManagerOptions> opts,
td::actor::ActorId<keyring::Keyring> keyring,
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp::Rldp> rldp,
td::actor::ActorId<dht::Dht> dht,
td::actor::ActorId<overlay::Overlays> overlays,
td::actor::ActorId<ValidatorManagerInterface> validator_manager,
td::actor::ActorId<adnl::AdnlExtClient> client, std::string db_root);
td::actor::ActorId<adnl::AdnlExtClient> client, std::string db_root,
td::Promise<td::Unit> started_promise);
};
} // namespace fullnode

27
validator/full-node.hpp
View file

@ -52,8 +52,7 @@ class FullNodeImpl : public FullNode {
void update_adnl_id(adnl::AdnlNodeIdShort adnl_id, td::Promise<td::Unit> promise) override;
void add_shard(ShardIdFull shard, bool subscribe = false);
void del_shard(ShardIdFull shard);
void update_shard_configuration(td::Ref<MasterchainState> state);
void sync_completed();
@ -82,21 +81,31 @@ class FullNodeImpl : public FullNode {
void start_up() override;
FullNodeImpl(PublicKeyHash local_id, adnl::AdnlNodeIdShort adnl_id, FileHash zero_state_file_hash,
td::actor::ActorId<keyring::Keyring> keyring, td::actor::ActorId<adnl::Adnl> adnl,
td::actor::ActorId<rldp::Rldp> rldp, td::actor::ActorId<dht::Dht> dht,
td::actor::ActorId<overlay::Overlays> overlays,
td::Ref<ValidatorManagerOptions> opts, td::actor::ActorId<keyring::Keyring> keyring,
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp::Rldp> rldp,
td::actor::ActorId<dht::Dht> dht, td::actor::ActorId<overlay::Overlays> overlays,
td::actor::ActorId<ValidatorManagerInterface> validator_manager,
td::actor::ActorId<adnl::AdnlExtClient> client, std::string db_root);
td::actor::ActorId<adnl::AdnlExtClient> client, std::string db_root,
td::Promise<td::Unit> started_promise);
private:
void add_shard_actor(ShardIdFull shard, bool active);
PublicKeyHash local_id_;
adnl::AdnlNodeIdShort adnl_id_;
FileHash zero_state_file_hash_;
td::Ref<ValidatorManagerOptions> opts_;
td::actor::ActorId<FullNodeShard> get_shard(AccountIdPrefixFull dst);
td::actor::ActorId<FullNodeShard> get_shard(ShardIdFull dst);
td::actor::ActorId<FullNodeShard> get_shard(ShardIdFull shard, bool exact = false);
std::map<ShardIdFull, td::actor::ActorOwn<FullNodeShard>> shards_;
struct ShardInfo {
bool exists = false;
td::actor::ActorOwn<FullNodeShard> actor;
bool active = false;
td::Timestamp delete_at = td::Timestamp::never();
};
std::map<ShardIdFull, ShardInfo> shards_;
td::actor::ActorId<keyring::Keyring> keyring_;
td::actor::ActorId<adnl::Adnl> adnl_;
@ -112,6 +121,8 @@ class FullNodeImpl : public FullNode {
std::vector<PublicKeyHash> all_validators_;
std::set<PublicKeyHash> local_keys_;
td::Promise<td::Unit> started_promise_;
};
} // namespace fullnode

42
validator/impl/accept-block.cpp
View file

@ -41,7 +41,7 @@ using namespace std::literals::string_literals;
AcceptBlockQuery::AcceptBlockQuery(BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::Ref<BlockSignatureSet> signatures,
td::Ref<BlockSignatureSet> approve_signatures, bool send_broadcast,
td::Ref<BlockSignatureSet> approve_signatures, bool send_broadcast, bool apply,
td::actor::ActorId<ValidatorManager> manager, td::Promise<td::Unit> promise)
: id_(id)
, data_(std::move(data))
@ -52,6 +52,7 @@ AcceptBlockQuery::AcceptBlockQuery(BlockIdExt id, td::Ref<BlockData> data, std::
, is_fake_(false)
, is_fork_(false)
, send_broadcast_(send_broadcast)
, apply_(apply)
, manager_(manager)
, promise_(std::move(promise)) {
state_keep_old_hash_.clear();
@ -92,28 +93,6 @@ AcceptBlockQuery::AcceptBlockQuery(ForceFork ffork, BlockIdExt id, td::Ref<Block
state_hash_.clear();
}
AcceptBlockQuery::AcceptBlockQuery(BroadcastOnly, BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::Ref<BlockSignatureSet> signatures,
td::Ref<BlockSignatureSet> approve_signatures,
td::actor::ActorId<ValidatorManager> manager, td::Promise<td::Unit> promise)
: id_(id)
, data_(std::move(data))
, prev_(std::move(prev))
, validator_set_(std::move(validator_set))
, signatures_(std::move(signatures))
, approve_signatures_(std::move(approve_signatures))
, is_fake_(false)
, is_fork_(false)
, send_broadcast_(true)
, broadcast_only_(false)
, manager_(manager)
, promise_(std::move(promise)) {
state_keep_old_hash_.clear();
state_old_hash_.clear();
state_hash_.clear();
CHECK(prev_.size() > 0);
}
bool AcceptBlockQuery::precheck_header() {
VLOG(VALIDATOR_DEBUG) << "precheck_header()";
// 0. sanity check
@ -357,7 +336,9 @@ bool AcceptBlockQuery::check_send_error(td::actor::ActorId<AcceptBlockQuery> Sel
}
void AcceptBlockQuery::finish_query() {
if (apply_) {
ValidatorInvariants::check_post_accept(handle_);
}
if (is_masterchain()) {
CHECK(handle_->inited_proof());
} else {
@ -406,15 +387,6 @@ void AcceptBlockQuery::start_up() {
return;
}
if (broadcast_only_) {
if (!create_new_proof()) {
fatal_error("cannot generate proof for block "s + id_.to_str());
return;
}
applied();
return;
}
td::actor::send_closure(
manager_, &ValidatorManager::get_block_handle, id_, true, [SelfId = actor_id(this)](td::Result<BlockHandle> R) {
check_send_error(SelfId, R) ||
@ -479,6 +451,10 @@ void AcceptBlockQuery::written_block_signatures() {
void AcceptBlockQuery::written_block_info() {
VLOG(VALIDATOR_DEBUG) << "written block info";
if (data_.not_null()) {
if (!apply_) {
written_state({});
return;
}
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::Ref<ShardState>> R) {
check_send_error(SelfId, R) ||
td::actor::send_closure_bool(SelfId, &AcceptBlockQuery::got_prev_state, R.move_as_ok());
@ -555,7 +531,7 @@ void AcceptBlockQuery::written_state(td::Ref<ShardState> upd_state) {
return;
}
if (state_keep_old_hash_ != state_old_hash_) {
if (apply_ && state_keep_old_hash_ != state_old_hash_) {
fatal_error(PSTRING() << "invalid previous state hash in newly-created proof: expected "
<< state_->root_hash().to_hex() << ", found in update " << state_old_hash_.to_hex());
return;

9
validator/impl/accept-block.hpp
View file

@ -48,20 +48,15 @@ class AcceptBlockQuery : public td::actor::Actor {
public:
struct IsFake {};
struct ForceFork {};
struct BroadcastOnly{};
AcceptBlockQuery(BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::Ref<BlockSignatureSet> signatures,
td::Ref<BlockSignatureSet> approve_signatures, bool send_broadcast,
td::Ref<BlockSignatureSet> approve_signatures, bool send_broadcast, bool apply,
td::actor::ActorId<ValidatorManager> manager, td::Promise<td::Unit> promise);
AcceptBlockQuery(IsFake fake, BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::actor::ActorId<ValidatorManager> manager,
td::Promise<td::Unit> promise);
AcceptBlockQuery(ForceFork ffork, BlockIdExt id, td::Ref<BlockData> data,
td::actor::ActorId<ValidatorManager> manager, td::Promise<td::Unit> promise);
AcceptBlockQuery(BroadcastOnly, BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::Ref<BlockSignatureSet> signatures,
td::Ref<BlockSignatureSet> approve_signatures, td::actor::ActorId<ValidatorManager> manager,
td::Promise<td::Unit> promise);
private:
static constexpr td::uint32 priority() {
@ -103,7 +98,7 @@ class AcceptBlockQuery : public td::actor::Actor {
bool is_fake_;
bool is_fork_;
bool send_broadcast_;
bool broadcast_only_{false};
bool apply_ = true;
bool ancestors_split_{false}, is_key_block_{false};
td::Timestamp timeout_ = td::Timestamp::in(600.0);
td::actor::ActorId<ValidatorManager> manager_;

15
validator/impl/fabric.cpp
View file

@ -127,10 +127,10 @@ td::Result<td::Ref<IhrMessage>> create_ihr_message(td::BufferSlice data) {
void run_accept_block_query(BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::Ref<BlockSignatureSet> signatures,
td::Ref<BlockSignatureSet> approve_signatures, bool send_broadcast,
td::Ref<BlockSignatureSet> approve_signatures, bool send_broadcast, bool apply,
td::actor::ActorId<ValidatorManager> manager, td::Promise<td::Unit> promise) {
td::actor::create_actor<AcceptBlockQuery>("accept", id, std::move(data), prev, std::move(validator_set),
std::move(signatures), std::move(approve_signatures), send_broadcast,
std::move(signatures), std::move(approve_signatures), send_broadcast, apply,
manager, std::move(promise))
.release();
}
@ -151,17 +151,6 @@ void run_hardfork_accept_block_query(BlockIdExt id, td::Ref<BlockData> data,
.release();
}
void run_broadcast_only_accept_block_query(BlockIdExt id, td::Ref<BlockData> data, std::vector<BlockIdExt> prev,
td::Ref<ValidatorSet> validator_set, td::Ref<BlockSignatureSet> signatures,
td::Ref<BlockSignatureSet> approve_signatures,
td::actor::ActorId<ValidatorManager> manager,
td::Promise<td::Unit> promise) {
td::actor::create_actor<AcceptBlockQuery>("broadcastaccept", AcceptBlockQuery::BroadcastOnly(), id, std::move(data),
prev, std::move(validator_set), std::move(signatures),
std::move(approve_signatures), manager, std::move(promise))
.release();
}
void run_apply_block_query(BlockIdExt id, td::Ref<BlockData> block, BlockIdExt masterchain_block_id,
td::actor::ActorId<ValidatorManager> manager, td::Timestamp timeout,
td::Promise<td::Unit> promise) {

6
validator/interfaces/validator-manager.h
View file

@ -64,10 +64,6 @@ class ValidatorManager : public ValidatorManagerInterface {
std::function<td::Status(td::FileFd&)> write_data,
td::Promise<td::Unit> promise) = 0;
virtual void store_zero_state_file(BlockIdExt block_id, td::BufferSlice state, td::Promise<td::Unit> promise) = 0;
virtual void wait_block_state(BlockHandle handle, td::uint32 priority, td::Timestamp timeout,
td::Promise<td::Ref<ShardState>> promise) = 0;
virtual void wait_block_state_short(BlockIdExt block_id, td::uint32 priority, td::Timestamp timeout,
td::Promise<td::Ref<ShardState>> promise) = 0;
virtual void set_block_data(BlockHandle handle, td::Ref<BlockData> data, td::Promise<td::Unit> promise) = 0;
virtual void wait_block_data(BlockHandle handle, td::uint32 priority, td::Timestamp,
@ -135,7 +131,7 @@ class ValidatorManager : public ValidatorManagerInterface {
virtual void update_shard_client_state(BlockIdExt masterchain_block_id, td::Promise<td::Unit> promise) = 0;
virtual void get_shard_client_state(bool from_db, td::Promise<BlockIdExt> promise) = 0;
virtual void subscribe_to_shard(ShardIdFull shard) = 0;
virtual void update_shard_configuration(td::Ref<MasterchainState> state) = 0;
virtual void update_async_serializer_state(AsyncSerializerState state, td::Promise<td::Unit> promise) = 0;
virtual void get_async_serializer_state(td::Promise<AsyncSerializerState> promise) = 0;

2
validator/manager-disk.hpp
View file

@ -253,7 +253,7 @@ class ValidatorManagerImpl : public ValidatorManager {
void update_shard_client_state(BlockIdExt masterchain_block_id, td::Promise<td::Unit> promise) override;
void get_shard_client_state(bool from_db, td::Promise<BlockIdExt> promise) override;
void subscribe_to_shard(ShardIdFull shard) override {
void update_shard_configuration(td::Ref<MasterchainState> state) override {
}
void update_async_serializer_state(AsyncSerializerState state, td::Promise<td::Unit> promise) override {

3
validator/manager-hardfork.hpp
View file

@ -321,7 +321,8 @@ class ValidatorManagerImpl : public ValidatorManager {
void get_shard_client_state(bool from_db, td::Promise<BlockIdExt> promise) override {
UNREACHABLE();
}
void subscribe_to_shard(ShardIdFull shard) override {
void update_shard_configuration(td::Ref<MasterchainState> state) override {
UNREACHABLE();
}
void update_async_serializer_state(AsyncSerializerState state, td::Promise<td::Unit> promise) override {

14
validator/manager.cpp
View file

@ -377,7 +377,7 @@ void ValidatorManagerImpl::new_external_message(td::BufferSlice data) {
}
auto R = create_ext_message(std::move(data));
if (R.is_error()) {
VLOG(VALIDATOR_NOTICE) << "dropping bad ihr message: " << R.move_as_error();
VLOG(VALIDATOR_NOTICE) << "dropping bad external message: " << R.move_as_error();
return;
}
add_external_message(R.move_as_ok());
@ -453,8 +453,8 @@ void ValidatorManagerImpl::add_shard_block_description(td::Ref<ShardTopBlockDesc
}
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) {
if (opts_->need_monitor(desc->block_id().shard_full()) && 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) {
@ -2367,8 +2367,8 @@ void ValidatorManagerImpl::get_shard_client_state(bool from_db, td::Promise<Bloc
}
}
void ValidatorManagerImpl::subscribe_to_shard(ShardIdFull shard) {
callback_->subscribe_to_shard(shard);
void ValidatorManagerImpl::update_shard_configuration(td::Ref<MasterchainState> state) {
callback_->update_shard_configuration(state);
}
void ValidatorManagerImpl::update_async_serializer_state(AsyncSerializerState state, td::Promise<td::Unit> promise) {
@ -2397,6 +2397,7 @@ void ValidatorManagerImpl::get_archive_slice(td::uint64 archive_id, td::uint64 o
}
bool ValidatorManagerImpl::is_validator() {
// TODO: change is_vaidator to other condition in some cases
return true; // temp_keys_.size() > 0 || permanent_keys_.size() > 0;
}
@ -2700,6 +2701,9 @@ void ValidatorManagerImpl::add_collator(adnl::AdnlNodeIdShort id, ShardIdFull sh
it = collator_nodes_.emplace(id, std::move(actor)).first;
}
td::actor::send_closure(it->second, &CollatorNode::add_shard, shard);
if (shard.is_masterchain()) {
collating_masterchain_ = true;
}
}
td::actor::ActorOwn<ValidatorManagerInterface> ValidatorManagerFactory::create(

3
validator/manager.hpp
View file

@ -447,7 +447,7 @@ class ValidatorManagerImpl : public ValidatorManager {
void update_shard_client_state(BlockIdExt masterchain_block_id, td::Promise<td::Unit> promise) override;
void get_shard_client_state(bool from_db, td::Promise<BlockIdExt> promise) override;
void subscribe_to_shard(ShardIdFull shard) override;
void update_shard_configuration(td::Ref<MasterchainState> state) override;
void update_async_serializer_state(AsyncSerializerState state, td::Promise<td::Unit> promise) override;
void get_async_serializer_state(td::Promise<AsyncSerializerState> promise) override;
@ -611,6 +611,7 @@ class ValidatorManagerImpl : public ValidatorManager {
void cleanup_old_pending_candidates(BlockId block_id, td::Timestamp now);
std::map<adnl::AdnlNodeIdShort, td::actor::ActorOwn<CollatorNode>> collator_nodes_;
bool collating_masterchain_ = false;
};
} // namespace validator

4
validator/net/download-block-new.cpp
View file

@ -201,10 +201,10 @@ void DownloadBlockNew::got_node_to_download(adnl::AdnlNodeIdShort node) {
}
if (client_.empty()) {
td::actor::send_closure(overlays_, &overlay::Overlays::send_query_via, download_from_, local_id_, overlay_id_,
"get_proof", std::move(P), td::Timestamp::in(3.0), std::move(q),
"get_block_full", std::move(P), td::Timestamp::in(3.0), std::move(q),
FullNode::max_proof_size() + FullNode::max_block_size() + 128, rldp_);
} else {
td::actor::send_closure(client_, &adnl::AdnlExtClient::send_query, "get_prepare",
td::actor::send_closure(client_, &adnl::AdnlExtClient::send_query, "get_block_full",
create_serialize_tl_object_suffix<ton_api::tonNode_query>(std::move(q)),
td::Timestamp::in(1.0), std::move(P));
}

34
validator/shard-client.cpp
View file

@ -247,39 +247,7 @@ void ShardClient::get_processed_masterchain_block_id(td::Promise<BlockIdExt> pro
}
void ShardClient::build_shard_overlays() {
auto v = masterchain_state_->get_shards();
std::set<WorkchainId> workchains;
for (auto &x : v) {
auto shard = x->shard();
workchains.insert(shard.workchain);
if (opts_->need_monitor(shard)) {
auto d = masterchain_state_->soft_min_split_depth(shard.workchain);
auto l = shard_prefix_length(shard.shard);
if (l > d) {
shard = shard_prefix(shard, d);
}
if (created_overlays_.count(shard) == 0) {
created_overlays_.insert(shard);
td::actor::send_closure(manager_, &ValidatorManager::subscribe_to_shard, shard);
}
}
}
for (const auto &wpair : masterchain_state_->get_workchain_list()) {
ton::WorkchainId wc = wpair.first;
const block::WorkchainInfo *winfo = wpair.second.get();
if (workchains.count(wc) == 0 && winfo->active && winfo->enabled_since <= masterchain_state_->get_unix_time()) {
auto shard = ShardIdFull(wc);
if (opts_->need_monitor(shard) && created_overlays_.count(shard) == 0) {
td::actor::send_closure(manager_, &ValidatorManager::subscribe_to_shard, shard);
BlockIdExt block_id(shard.workchain, shard.shard, 0, winfo->zerostate_root_hash, winfo->zerostate_file_hash);
td::actor::send_closure_later(manager_, &ValidatorManager::wait_block_state_short, block_id, 0,
td::Timestamp::in(5.0), [](td::Result<td::Ref<ShardState>>) {});
}
}
}
td::actor::send_closure(manager_, &ValidatorManager::update_shard_configuration, masterchain_state_);
}
void ShardClient::force_update_shard_client(BlockHandle handle, td::Promise<td::Unit> promise) {

16
validator/validator-group.cpp
View file

@ -105,8 +105,7 @@ void ValidatorGroup::validate_block_candidate(td::uint32 round_id, BlockCandidat
void ValidatorGroup::update_approve_cache(td::uint32 round_id, CacheKey key, UnixTime value) {
if (approved_candidates_cache_round_ != round_id) {
approved_candidates_cache_round_ = round_id;
approved_candidates_cache_.clear();
return;
}
approved_candidates_cache_[key] = value;
}
@ -157,15 +156,9 @@ void ValidatorGroup::accept_block_query(BlockIdExt block_id, td::Ref<BlockData>
}
});
if (shard_.is_masterchain() || lite_mode_) {
run_accept_block_query(block_id, std::move(block), std::move(prev), validator_set_, std::move(sig_set),
std::move(approve_sig_set), send_broadcast, manager_, std::move(P));
} else if (send_broadcast) {
run_broadcast_only_accept_block_query(block_id, std::move(block), std::move(prev), validator_set_,
std::move(sig_set), std::move(approve_sig_set), manager_, std::move(P));
} else {
promise.set_value(td::Unit());
}
std::move(approve_sig_set), send_broadcast, shard_.is_masterchain() || !lite_mode_, manager_,
std::move(P));
}
void ValidatorGroup::skip_round(td::uint32 round_id) {
@ -386,7 +379,7 @@ void ValidatorGroup::send_collate_query(td::uint32 round_id, td::Timestamp timeo
td::actor::send_closure(SelfId, &ValidatorGroup::receive_collate_query_response, round_id, R.move_as_ok(),
std::move(promise));
});
LOG(INFO) << "collate query for " << shard_.to_str() << ": send query to " << collator;
LOG(INFO) << "collate query for " << create_next_block_id_simple().to_str() << ": send query to " << collator;
size_t max_answer_size = config_.max_block_size + config_.max_collated_data_size + 256;
td::actor::send_closure(rldp_, &rldp::Rldp::send_query_ex, adnl::AdnlNodeIdShort(local_id_), collator, "collatequery",
std::move(P), timeout, std::move(query), max_answer_size);
@ -412,6 +405,7 @@ void ValidatorGroup::receive_collate_query_response(td::uint32 round_id, td::Buf
auto key = PublicKey{b->source_};
if (!key.is_ed25519()) {
promise.set_error(td::Status::Error("collate query: block candidate source mismatch"));
return;
}
auto e_key = Ed25519_PublicKey{key.ed25519_value().raw()};
if (e_key != Ed25519_PublicKey{local_id_full_.ed25519_value().raw()}) {

8
validator/validator.h
View file

@ -111,8 +111,7 @@ class ValidatorManagerInterface : public td::actor::Actor {
virtual ~Callback() = default;
virtual void initial_read_complete(BlockHandle top_masterchain_blocks) = 0;
virtual void subscribe_to_shard(ShardIdFull shard) = 0;
//virtual void del_shard(ShardIdFull shard) = 0;
virtual void update_shard_configuration(td::Ref<ton::validator::MasterchainState> state) = 0;
virtual void send_ihr_message(AccountIdPrefixFull dst, td::BufferSlice data) = 0;
virtual void send_ext_message(AccountIdPrefixFull dst, td::BufferSlice data) = 0;
@ -208,6 +207,11 @@ class ValidatorManagerInterface : public td::actor::Actor {
virtual void get_block_by_seqno_from_db(AccountIdPrefixFull account, BlockSeqno seqno,
td::Promise<ConstBlockHandle> promise) = 0;
virtual void wait_block_state(BlockHandle handle, td::uint32 priority, td::Timestamp timeout,
td::Promise<td::Ref<ShardState>> promise) = 0;
virtual void wait_block_state_short(BlockIdExt block_id, td::uint32 priority, td::Timestamp timeout,
td::Promise<td::Ref<ShardState>> promise) = 0;
virtual void get_archive_id(BlockSeqno masterchain_seqno, td::Promise<td::uint64> promise) = 0;
virtual void get_archive_slice(td::uint64 archive_id, td::uint64 offset, td::uint32 limit,
td::Promise<td::BufferSlice> promise) = 0;