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

Merge branch 'testnet' into block-generation

Browse source
This commit is contained in:
SpyCheese 2024-03-26 16:19:34 +03:00
commit 7999a7e2c1
52 changed files with 1466 additions and 282 deletions
Download patch file Download diff file Expand all files Collapse all files

2
validator/CMakeLists.txt
View file

@ -149,6 +149,8 @@ set(FULL_NODE_SOURCE
full-node-master.cpp
full-node-private-overlay.hpp
full-node-private-overlay.cpp
full-node-serializer.hpp
full-node-serializer.cpp
full-node-private-overlay-v2.hpp
full-node-private-overlay-v2.cpp

3
validator/full-node-master.cpp
View file

@ -371,7 +371,8 @@ void FullNodeMasterImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::tonNo
void FullNodeMasterImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::tonNode_getCapabilities &query,
td::Promise<td::BufferSlice> promise) {
promise.set_value(create_serialize_tl_object<ton_api::tonNode_capabilities>(proto_version(), proto_capabilities()));
promise.set_value(
create_serialize_tl_object<ton_api::tonNode_capabilities>(proto_version_major(), proto_version_minor(), 0));
}
void FullNodeMasterImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::tonNode_getArchiveInfo &query,

4
validator/full-node-master.hpp
View file

@ -28,10 +28,10 @@ namespace fullnode {
class FullNodeMasterImpl : public FullNodeMaster {
public:
static constexpr td::uint32 proto_version() {
static constexpr td::uint32 proto_version_major() {
return 1;
}
static constexpr td::uint64 proto_capabilities() {
static constexpr td::uint32 proto_version_minor() {
return 0;
}
void start_up() override;

43
validator/full-node-private-overlay.cpp
View file

@ -14,11 +14,10 @@
You should have received a copy of the GNU Lesser General Public License
along with TON Blockchain Library. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include "full-node-private-overlay.hpp"
#include "ton/ton-tl.hpp"
#include "common/delay.h"
#include "full-node-serializer.hpp"
namespace ton {
@ -26,20 +25,20 @@ namespace validator {
namespace fullnode {
void FullNodePrivateOverlay::process_broadcast(PublicKeyHash, ton_api::tonNode_blockBroadcast &query) {
std::vector<BlockSignature> signatures;
for (auto &sig : query.signatures_) {
signatures.emplace_back(BlockSignature{sig->who_, std::move(sig->signature_)});
void FullNodePrivateOverlay::process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcast &query) {
process_block_broadcast(src, query);
}
void FullNodePrivateOverlay::process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcastCompressed &query) {
process_block_broadcast(src, query);
}
void FullNodePrivateOverlay::process_block_broadcast(PublicKeyHash src, ton_api::tonNode_Broadcast &query) {
auto B = deserialize_block_broadcast(query, overlay::Overlays::max_fec_broadcast_size());
if (B.is_error()) {
LOG(DEBUG) << "dropped broadcast: " << B.move_as_error();
return;
}
BlockIdExt block_id = create_block_id(query.id_);
BlockBroadcast B{block_id,
std::move(signatures),
static_cast<UnixTime>(query.catchain_seqno_),
static_cast<td::uint32>(query.validator_set_hash_),
std::move(query.data_),
std::move(query.proof_)};
auto P = td::PromiseCreator::lambda([](td::Result<td::Unit> R) {
if (R.is_error()) {
if (R.error().code() == ErrorCode::notready) {
@ -49,7 +48,7 @@ void FullNodePrivateOverlay::process_broadcast(PublicKeyHash, ton_api::tonNode_b
}
}
});
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::prevalidate_block, std::move(B),
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::prevalidate_block, B.move_as_ok(),
std::move(P));
}
@ -87,15 +86,13 @@ void FullNodePrivateOverlay::send_broadcast(BlockBroadcast broadcast) {
if (!inited_) {
return;
}
std::vector<tl_object_ptr<ton_api::tonNode_blockSignature>> sigs;
for (auto &sig : broadcast.signatures) {
sigs.emplace_back(create_tl_object<ton_api::tonNode_blockSignature>(sig.node, sig.signature.clone()));
auto B = serialize_block_broadcast(broadcast, false); // compression_enabled = false
if (B.is_error()) {
VLOG(FULL_NODE_WARNING) << "failed to serialize block broadcast: " << B.move_as_error();
return;
}
auto B = create_serialize_tl_object<ton_api::tonNode_blockBroadcast>(
create_tl_block_id(broadcast.block_id), broadcast.catchain_seqno, broadcast.validator_set_hash, std::move(sigs),
broadcast.proof.clone(), broadcast.data.clone());
td::actor::send_closure(overlays_, &overlay::Overlays::send_broadcast_fec_ex, local_id_, overlay_id_,
local_id_.pubkey_hash(), overlay::Overlays::BroadcastFlagAnySender(), std::move(B));
local_id_.pubkey_hash(), overlay::Overlays::BroadcastFlagAnySender(), B.move_as_ok());
}
void FullNodePrivateOverlay::start_up() {

3
validator/full-node-private-overlay.hpp
View file

@ -27,6 +27,9 @@ namespace fullnode {
class FullNodePrivateOverlay : public td::actor::Actor {
public:
void process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcast &query);
void process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcastCompressed &query);
void process_block_broadcast(PublicKeyHash src, ton_api::tonNode_Broadcast &query);
void process_broadcast(PublicKeyHash src, ton_api::tonNode_newShardBlockBroadcast &query);
template <class T>
void process_broadcast(PublicKeyHash, T &) {

155
validator/full-node-serializer.cpp Normal file
View file

@ -0,0 +1,155 @@
/*
This file is part of TON Blockchain Library.
TON Blockchain Library is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 2 of the License, or
(at your option) any later version.
TON Blockchain Library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with TON Blockchain Library. If not, see <http://www.gnu.org/licenses/>.
*/
#include "full-node-serializer.hpp"
#include "ton/ton-tl.hpp"
#include "tl-utils/common-utils.hpp"
#include "auto/tl/ton_api.hpp"
#include "tl-utils/tl-utils.hpp"
#include "vm/boc.h"
#include "td/utils/lz4.h"
#include "full-node.h"
#include "td/utils/overloaded.h"
namespace ton::validator::fullnode {
td::Result<td::BufferSlice> serialize_block_broadcast(const BlockBroadcast& broadcast, bool compression_enabled) {
std::vector<tl_object_ptr<ton_api::tonNode_blockSignature>> sigs;
for (auto& sig : broadcast.signatures) {
sigs.emplace_back(create_tl_object<ton_api::tonNode_blockSignature>(sig.node, sig.signature.clone()));
}
if (!compression_enabled) {
return create_serialize_tl_object<ton_api::tonNode_blockBroadcast>(
create_tl_block_id(broadcast.block_id), broadcast.catchain_seqno, broadcast.validator_set_hash, std::move(sigs),
broadcast.proof.clone(), broadcast.data.clone());
}
TRY_RESULT(proof_root, vm::std_boc_deserialize(broadcast.proof));
TRY_RESULT(data_root, vm::std_boc_deserialize(broadcast.data));
TRY_RESULT(boc, vm::std_boc_serialize_multi({proof_root, data_root}, 2));
td::BufferSlice data =
create_serialize_tl_object<ton_api::tonNode_blockBroadcastCompressed_data>(std::move(sigs), std::move(boc));
td::BufferSlice compressed = td::lz4_compress(data);
VLOG(FULL_NODE_DEBUG) << "Compressing block broadcast: "
<< broadcast.data.size() + broadcast.proof.size() + broadcast.signatures.size() * 96 << " -> "
<< compressed.size();
return create_serialize_tl_object<ton_api::tonNode_blockBroadcastCompressed>(
create_tl_block_id(broadcast.block_id), broadcast.catchain_seqno, broadcast.validator_set_hash, 0,
std::move(compressed));
}
static td::Result<BlockBroadcast> deserialize_block_broadcast(ton_api::tonNode_blockBroadcast& f) {
std::vector<BlockSignature> signatures;
for (auto& sig : f.signatures_) {
signatures.emplace_back(BlockSignature{sig->who_, std::move(sig->signature_)});
}
return BlockBroadcast{create_block_id(f.id_),
std::move(signatures),
static_cast<UnixTime>(f.catchain_seqno_),
static_cast<td::uint32>(f.validator_set_hash_),
std::move(f.data_),
std::move(f.proof_)};
}
static td::Result<BlockBroadcast> deserialize_block_broadcast(ton_api::tonNode_blockBroadcastCompressed& f,
int max_decompressed_size) {
TRY_RESULT(decompressed, td::lz4_decompress(f.compressed_, max_decompressed_size));
TRY_RESULT(f2, fetch_tl_object<ton_api::tonNode_blockBroadcastCompressed_data>(decompressed, true));
std::vector<BlockSignature> signatures;
for (auto& sig : f2->signatures_) {
signatures.emplace_back(BlockSignature{sig->who_, std::move(sig->signature_)});
}
TRY_RESULT(roots, vm::std_boc_deserialize_multi(f2->proof_data_, 2));
if (roots.size() != 2) {
return td::Status::Error("expected 2 roots in boc");
}
TRY_RESULT(proof, vm::std_boc_serialize(roots[0], 0));
TRY_RESULT(data, vm::std_boc_serialize(roots[1], 31));
VLOG(FULL_NODE_DEBUG) << "Decompressing block broadcast: " << f.compressed_.size() << " -> "
<< data.size() + proof.size() + signatures.size() * 96;
return BlockBroadcast{create_block_id(f.id_),
std::move(signatures),
static_cast<UnixTime>(f.catchain_seqno_),
static_cast<td::uint32>(f.validator_set_hash_),
std::move(data),
std::move(proof)};
}
td::Result<BlockBroadcast> deserialize_block_broadcast(ton_api::tonNode_Broadcast& obj,
int max_decompressed_data_size) {
td::Result<BlockBroadcast> B;
ton_api::downcast_call(obj,
td::overloaded([&](ton_api::tonNode_blockBroadcast& f) { B = deserialize_block_broadcast(f); },
[&](ton_api::tonNode_blockBroadcastCompressed& f) {
B = deserialize_block_broadcast(f, max_decompressed_data_size);
},
[&](auto&) { B = td::Status::Error("unknown broadcast type"); }));
return B;
}
td::Result<td::BufferSlice> serialize_block_full(const BlockIdExt& id, td::Slice proof, td::Slice data,
bool is_proof_link, bool compression_enabled) {
if (!compression_enabled) {
return create_serialize_tl_object<ton_api::tonNode_dataFull>(create_tl_block_id(id), td::BufferSlice(proof),
td::BufferSlice(data), is_proof_link);
}
TRY_RESULT(proof_root, vm::std_boc_deserialize(proof));
TRY_RESULT(data_root, vm::std_boc_deserialize(data));
TRY_RESULT(boc, vm::std_boc_serialize_multi({proof_root, data_root}, 2));
td::BufferSlice compressed = td::lz4_compress(boc);
VLOG(FULL_NODE_DEBUG) << "Compressing block full: " << data.size() + proof.size() << " -> " << compressed.size();
return create_serialize_tl_object<ton_api::tonNode_dataFullCompressed>(create_tl_block_id(id), 0,
std::move(compressed), is_proof_link);
}
static td::Status deserialize_block_full(ton_api::tonNode_dataFull& f, BlockIdExt& id, td::BufferSlice& proof,
td::BufferSlice& data, bool& is_proof_link) {
id = create_block_id(f.id_);
proof = std::move(f.proof_);
data = std::move(f.block_);
is_proof_link = f.is_link_;
return td::Status::OK();
}
static td::Status deserialize_block_full(ton_api::tonNode_dataFullCompressed& f, BlockIdExt& id, td::BufferSlice& proof,
td::BufferSlice& data, bool& is_proof_link, int max_decompressed_size) {
TRY_RESULT(decompressed, td::lz4_decompress(f.compressed_, max_decompressed_size));
TRY_RESULT(roots, vm::std_boc_deserialize_multi(decompressed, 2));
if (roots.size() != 2) {
return td::Status::Error("expected 2 roots in boc");
}
TRY_RESULT_ASSIGN(proof, vm::std_boc_serialize(roots[0], 0));
TRY_RESULT_ASSIGN(data, vm::std_boc_serialize(roots[1], 31));
VLOG(FULL_NODE_DEBUG) << "Decompressing block full: " << f.compressed_.size() << " -> " << data.size() + proof.size();
id = create_block_id(f.id_);
is_proof_link = f.is_link_;
return td::Status::OK();
}
td::Status deserialize_block_full(ton_api::tonNode_DataFull& obj, BlockIdExt& id, td::BufferSlice& proof,
td::BufferSlice& data, bool& is_proof_link, int max_decompressed_data_size) {
td::Status S;
ton_api::downcast_call(
obj, td::overloaded(
[&](ton_api::tonNode_dataFull& f) { S = deserialize_block_full(f, id, proof, data, is_proof_link); },
[&](ton_api::tonNode_dataFullCompressed& f) {
S = deserialize_block_full(f, id, proof, data, is_proof_link, max_decompressed_data_size);
},
[&](auto&) { S = td::Status::Error("unknown data type"); }));
return S;
}
} // namespace ton::validator::fullnode

31
validator/full-node-serializer.hpp Normal file
View file

@ -0,0 +1,31 @@
/*
This file is part of TON Blockchain Library.
TON Blockchain Library is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 2 of the License, or
(at your option) any later version.
TON Blockchain Library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with TON Blockchain Library. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include "ton/ton-types.h"
#include "auto/tl/ton_api.h"
namespace ton::validator::fullnode {
td::Result<td::BufferSlice> serialize_block_broadcast(const BlockBroadcast& broadcast, bool compression_enabled);
td::Result<BlockBroadcast> deserialize_block_broadcast(ton_api::tonNode_Broadcast& obj, int max_decompressed_data_size);
td::Result<td::BufferSlice> serialize_block_full(const BlockIdExt& id, td::Slice proof, td::Slice data,
bool is_proof_link, bool compression_enabled);
td::Status deserialize_block_full(ton_api::tonNode_DataFull& obj, BlockIdExt& id, td::BufferSlice& proof,
td::BufferSlice& data, bool& is_proof_link, int max_decompressed_data_size);
} // namespace ton::validator::fullnode

5
validator/full-node-shard-queries.hpp
View file

@ -20,6 +20,7 @@
#include "validator/validator.h"
#include "ton/ton-tl.hpp"
#include "full-node-serializer.hpp"
namespace ton {
@ -38,8 +39,8 @@ class BlockFullSender : public td::actor::Actor {
stop();
}
void finish_query() {
promise_.set_value(create_serialize_tl_object<ton_api::tonNode_dataFull>(
create_tl_block_id(block_id_), std::move(proof_), std::move(data_), is_proof_link_));
promise_.set_result(
serialize_block_full(block_id_, proof_, data_, is_proof_link_, false)); // compression_enabled = false
stop();
}
void start_up() override {

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

@ -22,6 +22,7 @@
#include "td/utils/overloaded.h"
#include "full-node-shard.hpp"
#include "full-node-shard-queries.hpp"
#include "full-node-serializer.hpp"
#include "ton/ton-shard.h"
#include "ton/ton-tl.hpp"
@ -739,24 +740,24 @@ void FullNodeShardImpl::process_broadcast(PublicKeyHash src, ton_api::tonNode_ne
}
void FullNodeShardImpl::process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcast &query) {
BlockIdExt block_id = create_block_id(query.id_);
process_block_broadcast(src, query);
}
void FullNodeShardImpl::process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcastCompressed &query) {
process_block_broadcast(src, query);
}
void FullNodeShardImpl::process_block_broadcast(PublicKeyHash src, ton_api::tonNode_Broadcast &query) {
auto B = deserialize_block_broadcast(query, overlay::Overlays::max_fec_broadcast_size());
if (B.is_error()) {
LOG(DEBUG) << "dropped broadcast: " << B.move_as_error();
return;
}
//if (!shard_is_ancestor(shard_, block_id.shard_full())) {
// 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_)});
}
BlockBroadcast B{block_id,
std::move(signatures),
static_cast<UnixTime>(query.catchain_seqno_),
static_cast<td::uint32>(query.validator_set_hash_),
std::move(query.data_),
std::move(query.proof_)};
auto P = td::PromiseCreator::lambda([](td::Result<td::Unit> R) {
if (R.is_error()) {
if (R.error().code() == ErrorCode::notready) {
@ -766,7 +767,7 @@ void FullNodeShardImpl::process_broadcast(PublicKeyHash src, ton_api::tonNode_bl
}
}
});
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::prevalidate_block, std::move(B),
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::prevalidate_block, B.move_as_ok(),
std::move(P));
}
@ -851,15 +852,13 @@ void FullNodeShardImpl::send_broadcast(BlockBroadcast broadcast) {
UNREACHABLE();
return;
}
std::vector<tl_object_ptr<ton_api::tonNode_blockSignature>> sigs;
for (auto &sig : broadcast.signatures) {
sigs.emplace_back(create_tl_object<ton_api::tonNode_blockSignature>(sig.node, sig.signature.clone()));
auto B = serialize_block_broadcast(broadcast, false); // compression_enabled = false
if (B.is_error()) {
VLOG(FULL_NODE_WARNING) << "failed to serialize block broadcast: " << B.move_as_error();
return;
}
auto B = create_serialize_tl_object<ton_api::tonNode_blockBroadcast>(
create_tl_block_id(broadcast.block_id), broadcast.catchain_seqno, broadcast.validator_set_hash, std::move(sigs),
broadcast.proof.clone(), broadcast.data.clone());
td::actor::send_closure(overlays_, &overlay::Overlays::send_broadcast_fec_ex, adnl_id_, overlay_id_, local_id_,
overlay::Overlays::BroadcastFlagAnySender(), std::move(B));
overlay::Overlays::BroadcastFlagAnySender(), B.move_as_ok());
}
void FullNodeShardImpl::download_block(BlockIdExt id, td::uint32 priority, td::Timestamp timeout,

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

@ -161,6 +161,9 @@ class FullNodeShardImpl : public FullNodeShard {
void receive_message(adnl::AdnlNodeIdShort src, td::BufferSlice data);
void process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcast &query);
void process_broadcast(PublicKeyHash src, ton_api::tonNode_blockBroadcastCompressed &query);
void process_block_broadcast(PublicKeyHash src, ton_api::tonNode_Broadcast &query);
void process_broadcast(PublicKeyHash src, ton_api::tonNode_ihrMessageBroadcast &query);
void process_broadcast(PublicKeyHash src, ton_api::tonNode_externalMessageBroadcast &query);
void process_broadcast(PublicKeyHash src, ton_api::tonNode_newShardBlockBroadcast &query);

2
validator/impl/CMakeLists.txt
View file

@ -7,6 +7,7 @@ endif()
set(TON_VALIDATOR_SOURCE
accept-block.cpp
block.cpp
candidates-buffer.cpp
check-proof.cpp
collator.cpp
config.cpp
@ -25,6 +26,7 @@ set(TON_VALIDATOR_SOURCE
accept-block.hpp
block.hpp
candidates-buffer.hpp
check-proof.hpp
collator-impl.h
collator.h

213
validator/impl/candidates-buffer.cpp Normal file
View file

@ -0,0 +1,213 @@
/*
This file is part of TON Blockchain Library.
TON Blockchain Library is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 2 of the License, or
(at your option) any later version.
TON Blockchain Library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with TON Blockchain Library. If not, see <http://www.gnu.org/licenses/>.
*/
#include "candidates-buffer.hpp"
#include "fabric.h"
namespace ton::validator {
void CandidatesBuffer::start_up() {
alarm_timestamp() = td::Timestamp::in(60.0);
}
void CandidatesBuffer::alarm() {
alarm_timestamp() = td::Timestamp::in(60.0);
for (auto it = candidates_.begin(); it != candidates_.end();) {
Candidate &entry = it->second;
if (entry.ttl_.is_in_past()) {
for (auto &p : entry.data_waiters_) {
p.set_error(td::Status::Error(ErrorCode::timeout, "timeout"));
}
for (auto &p : entry.state_waiters_) {
p.set_error(td::Status::Error(ErrorCode::timeout, "timeout"));
}
it = candidates_.erase(it);
} else {
++it;
}
}
}
void CandidatesBuffer::add_new_candidate(BlockIdExt id, PublicKey source, FileHash collated_data_file_hash) {
auto it = candidates_.emplace(id, Candidate{});
Candidate &entry = it.first->second;
entry.ttl_ = td::Timestamp::in(120.0);
if (!it.second) { // not inserted
return;
}
LOG(DEBUG) << "New block candidate " << id.to_str();
entry.source_ = source;
entry.collated_data_file_hash_ = collated_data_file_hash;
}
void CandidatesBuffer::get_block_data(BlockIdExt id, td::Promise<td::Ref<BlockData>> promise) {
auto it = candidates_.find(id);
if (it == candidates_.end()) {
promise.set_error(td::Status::Error(ErrorCode::notready, "unknown block candidate"));
return;
}
Candidate &entry = it->second;
if (entry.data_.not_null()) {
promise.set_result(entry.data_);
return;
}
entry.data_waiters_.push_back(std::move(promise));
if (entry.data_requested_) {
return;
}
entry.data_requested_ = true;
td::actor::send_closure(manager_, &ValidatorManager::get_block_candidate_from_db, entry.source_, id,
entry.collated_data_file_hash_, [SelfId = actor_id(this), id](td::Result<BlockCandidate> R) {
td::actor::send_closure(SelfId, &CandidatesBuffer::got_block_candidate, id, std::move(R));
});
}
void CandidatesBuffer::got_block_candidate(BlockIdExt id, td::Result<BlockCandidate> R) {
if (R.is_error()) {
finish_get_block_data(id, R.move_as_error());
return;
}
BlockCandidate cand = R.move_as_ok();
CHECK(cand.id == id);
finish_get_block_data(id, create_block(id, std::move(cand.data)));
}
void CandidatesBuffer::get_block_state(BlockIdExt id, td::Promise<td::Ref<ShardState>> promise) {
auto it = candidates_.find(id);
if (it == candidates_.end()) {
promise.set_error(td::Status::Error(ErrorCode::notready, "unknown block candidate"));
return;
}
Candidate &entry = it->second;
if (entry.state_.not_null()) {
promise.set_result(entry.state_);
return;
}
entry.state_waiters_.push_back(std::move(promise));
if (entry.state_requested_) {
return;
}
entry.state_requested_ = true;
get_block_data(id, [SelfId = actor_id(this), id](td::Result<td::Ref<BlockData>> R) {
if (R.is_error()) {
td::actor::send_closure(SelfId, &CandidatesBuffer::finish_get_block_state, id, R.move_as_error());
return;
}
td::actor::send_closure(SelfId, &CandidatesBuffer::get_block_state_cont, id, R.move_as_ok());
});
}
void CandidatesBuffer::get_block_state_cont(BlockIdExt id, td::Ref<BlockData> data) {
CHECK(id == data->block_id());
std::vector<BlockIdExt> prev;
BlockIdExt mc_blkid;
bool after_split;
auto S = block::unpack_block_prev_blk_ext(data->root_cell(), id, prev, mc_blkid, after_split);
if (S.is_error()) {
finish_get_block_state(id, std::move(S));
return;
}
get_block_state_cont2(std::move(data), std::move(prev), {});
}
void CandidatesBuffer::get_block_state_cont2(td::Ref<BlockData> block, std::vector<BlockIdExt> prev,
std::vector<td::Ref<ShardState>> prev_states) {
if (prev_states.size() < prev.size()) {
BlockIdExt prev_id = prev[prev_states.size()];
td::actor::send_closure(manager_, &ValidatorManager::get_shard_state_from_db_short, prev_id,
[SelfId = actor_id(this), block = std::move(block), prev = std::move(prev),
prev_states = std::move(prev_states)](td::Result<td::Ref<ShardState>> R) mutable {
if (R.is_error()) {
td::actor::send_closure(SelfId, &CandidatesBuffer::finish_get_block_state,
block->block_id(), R.move_as_error());
return;
}
prev_states.push_back(R.move_as_ok());
td::actor::send_closure(SelfId, &CandidatesBuffer::get_block_state_cont2,
std::move(block), std::move(prev), std::move(prev_states));
});
return;
}
BlockIdExt id = block->block_id();
td::Ref<ShardState> state;
CHECK(prev_states.size() == 1 || prev_states.size() == 2);
if (prev_states.size() == 2) { // after merge
auto R = prev_states[0]->merge_with(*prev_states[1]);
if (R.is_error()) {
finish_get_block_state(id, R.move_as_error());
return;
}
state = R.move_as_ok();
} else if (id.shard_full() != prev[0].shard_full()) { // after split
auto R = prev_states[0]->split();
if (R.is_error()) {
finish_get_block_state(id, R.move_as_error());
return;
}
auto s = R.move_as_ok();
state = is_left_child(id.shard_full()) ? std::move(s.first) : std::move(s.second);
} else { // no split/merge
state = std::move(prev_states[0]);
}
auto S = state.write().apply_block(id, std::move(block));
if (S.is_error()) {
finish_get_block_state(id, std::move(S));
return;
}
finish_get_block_state(id, std::move(state));
}
void CandidatesBuffer::finish_get_block_data(BlockIdExt id, td::Result<td::Ref<BlockData>> res) {
auto it = candidates_.find(id);
if (it == candidates_.end()) {
return;
}
Candidate &entry = it->second;
for (auto &p : entry.data_waiters_) {
p.set_result(res.clone());
}
entry.data_waiters_.clear();
entry.data_requested_ = false;
if (res.is_ok()) {
entry.data_ = res.move_as_ok();
LOG(DEBUG) << "Loaded block data for " << id.to_str();
} else {
LOG(DEBUG) << "Failed to load block data for " << id.to_str() << ": " << res.move_as_error();
}
}
void CandidatesBuffer::finish_get_block_state(BlockIdExt id, td::Result<td::Ref<ShardState>> res) {
auto it = candidates_.find(id);
if (it == candidates_.end()) {
return;
}
Candidate &entry = it->second;
for (auto &p : entry.state_waiters_) {
p.set_result(res.clone());
}
entry.state_waiters_.clear();
entry.state_requested_ = false;
if (res.is_ok()) {
entry.state_ = res.move_as_ok();
LOG(DEBUG) << "Loaded block state for " << id.to_str();
} else {
LOG(DEBUG) << "Failed to load block state for " << id.to_str() << ": " << res.move_as_error();
}
}
} // namespace ton::validator

64
validator/impl/candidates-buffer.hpp Normal file
View file

@ -0,0 +1,64 @@
/*
This file is part of TON Blockchain Library.
TON Blockchain Library is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 2 of the License, or
(at your option) any later version.
TON Blockchain Library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with TON Blockchain Library. If not, see <http://www.gnu.org/licenses/>.
*/
#pragma once
#include "ton/ton-types.h"
#include "td/actor/actor.h"
#include "interfaces/validator-manager.h"
namespace ton::validator {
class CandidatesBuffer : public td::actor::Actor {
public:
explicit CandidatesBuffer(td::actor::ActorId<ValidatorManager> manager) : manager_(std::move(manager)) {
}
void start_up() override;
void alarm() override;
void add_new_candidate(BlockIdExt id, PublicKey source, FileHash collated_data_file_hash);
void get_block_data(BlockIdExt id, td::Promise<td::Ref<BlockData>> promise);
void get_block_state(BlockIdExt id, td::Promise<td::Ref<ShardState>> promise);
private:
td::actor::ActorId<ValidatorManager> manager_;
struct Candidate {
PublicKey source_;
FileHash collated_data_file_hash_;
td::Timestamp ttl_;
td::Ref<BlockData> data_;
std::vector<td::Promise<td::Ref<BlockData>>> data_waiters_;
bool data_requested_{false};
td::Ref<ShardState> state_;
std::vector<td::Promise<td::Ref<ShardState>>> state_waiters_;
bool state_requested_{false};
};
std::map<BlockIdExt, Candidate> candidates_;
void got_block_candidate(BlockIdExt id, td::Result<BlockCandidate> R);
void get_block_state_cont(BlockIdExt id, td::Ref<BlockData> data);
void get_block_state_cont2(td::Ref<BlockData> block, std::vector<BlockIdExt> prev,
std::vector<td::Ref<ShardState>> prev_states);
void finish_get_block_data(BlockIdExt id, td::Result<td::Ref<BlockData>> res);
void finish_get_block_state(BlockIdExt id, td::Result<td::Ref<ShardState>> res);
};
} // namespace ton::validator

3
validator/impl/collator-impl.h
View file

@ -336,6 +336,9 @@ class Collator final : public td::actor::Actor {
bool create_block_candidate();
void return_block_candidate(td::Result<td::Unit> saved);
bool update_last_proc_int_msg(const std::pair<ton::LogicalTime, ton::Bits256>& new_lt_hash);
public:
static td::uint32 get_skip_externals_queue_size();
};
} // namespace validator

4
validator/impl/collator.cpp
View file

@ -5352,6 +5352,10 @@ void Collator::after_get_external_messages(td::Result<std::vector<Ref<ExtMessage
check_pending();
}
td::uint32 Collator::get_skip_externals_queue_size() {
return SKIP_EXTERNALS_QUEUE_SIZE;
}
} // namespace validator
} // namespace ton

269
validator/impl/liteserver.cpp
View file

@ -42,6 +42,8 @@
#include "signature-set.hpp"
#include "fabric.h"
#include <ctime>
#include "td/actor/MultiPromise.h"
#include "collator-impl.h"
namespace ton {
@ -275,6 +277,16 @@ void LiteQuery::perform() {
[&](lite_api::liteServer_getShardBlockProof& q) {
this->perform_getShardBlockProof(create_block_id(q.id_));
},
[&](lite_api::liteServer_nonfinal_getCandidate& q) {
this->perform_nonfinal_getCandidate(q.id_->creator_, create_block_id(q.id_->block_id_),
q.id_->collated_data_hash_);
},
[&](lite_api::liteServer_nonfinal_getValidatorGroups& q) {
this->perform_nonfinal_getValidatorGroups(q.mode_, ShardIdFull{q.wc_, (ShardId)q.shard_});
},
[&](lite_api::liteServer_getOutMsgQueueSizes& q) {
this->perform_getOutMsgQueueSizes(q.mode_ & 1 ? ShardIdFull(q.wc_, q.shard_) : td::optional<ShardIdFull>());
},
[&](auto& obj) { this->abort_query(td::Status::Error(ErrorCode::protoviolation, "unknown query")); }));
}
@ -343,21 +355,15 @@ void LiteQuery::perform_getBlock(BlockIdExt blkid) {
fatal_error("invalid BlockIdExt");
return;
}
get_block_handle_checked(blkid, [manager = manager_, Self = actor_id(this), blkid](td::Result<ConstBlockHandle> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
return;
}
td::actor::send_closure_later(manager, &ValidatorManager::get_block_data_from_db, R.move_as_ok(),
[=](td::Result<Ref<ton::validator::BlockData>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getBlock, blkid,
res.move_as_ok());
}
});
});
td::actor::send_closure(manager_, &ValidatorManager::get_block_data_for_litequery, blkid,
[Self = actor_id(this), blkid](td::Result<Ref<ton::validator::BlockData>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getBlock, blkid,
res.move_as_ok());
}
});
}
void LiteQuery::continue_getBlock(BlockIdExt blkid, Ref<ton::validator::BlockData> block) {
@ -375,21 +381,15 @@ void LiteQuery::perform_getBlockHeader(BlockIdExt blkid, int mode) {
fatal_error("invalid BlockIdExt");
return;
}
get_block_handle_checked(blkid, [=, manager = manager_, Self = actor_id(this)](td::Result<ConstBlockHandle> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
return;
}
td::actor::send_closure_later(manager, &ValidatorManager::get_block_data_from_db, R.move_as_ok(),
[=](td::Result<Ref<ton::validator::BlockData>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getBlockHeader, blkid,
mode, res.move_as_ok());
}
});
});
td::actor::send_closure(manager_, &ValidatorManager::get_block_data_for_litequery, blkid,
[Self = actor_id(this), blkid, mode](td::Result<Ref<ton::validator::BlockData>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getBlockHeader, blkid, mode,
res.move_as_ok());
}
});
}
static bool visit(Ref<vm::Cell> cell);
@ -495,33 +495,27 @@ void LiteQuery::perform_getState(BlockIdExt blkid) {
fatal_error("cannot request total state: possibly too large");
return;
}
get_block_handle_checked(blkid, [=, manager = manager_, Self = actor_id(this)](td::Result<ConstBlockHandle> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
return;
}
if (blkid.id.seqno) {
td::actor::send_closure_later(manager, &ValidatorManager::get_shard_state_from_db, R.move_as_ok(),
[=](td::Result<Ref<ton::validator::ShardState>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getState, blkid,
res.move_as_ok());
}
});
} else {
td::actor::send_closure_later(manager, &ValidatorManager::get_zero_state, blkid,
[=](td::Result<td::BufferSlice> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getZeroState, blkid,
res.move_as_ok());
}
});
}
});
if (blkid.id.seqno) {
td::actor::send_closure(manager_, &ValidatorManager::get_block_state_for_litequery, blkid,
[Self = actor_id(this), blkid](td::Result<Ref<ShardState>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getState, blkid,
res.move_as_ok());
}
});
} else {
td::actor::send_closure_later(manager_, &ValidatorManager::get_zero_state, blkid,
[Self = actor_id(this), blkid](td::Result<td::BufferSlice> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getZeroState, blkid,
res.move_as_ok());
}
});
}
}
void LiteQuery::continue_getState(BlockIdExt blkid, Ref<ton::validator::ShardState> state) {
@ -583,7 +577,7 @@ bool LiteQuery::request_mc_block_data(BlockIdExt blkid) {
base_blk_id_ = blkid;
++pending_;
td::actor::send_closure_later(
manager_, &ValidatorManager::get_block_data_from_db_short, blkid,
manager_, &ValidatorManager::get_block_data_for_litequery, blkid,
[Self = actor_id(this), blkid](td::Result<Ref<BlockData>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query,
@ -641,7 +635,7 @@ bool LiteQuery::request_mc_block_state(BlockIdExt blkid) {
base_blk_id_ = blkid;
++pending_;
td::actor::send_closure_later(
manager_, &ValidatorManager::get_shard_state_from_db_short, blkid,
manager_, &ValidatorManager::get_block_state_for_litequery, blkid,
[Self = actor_id(this), blkid](td::Result<Ref<ShardState>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query,
@ -671,22 +665,16 @@ bool LiteQuery::request_block_state(BlockIdExt blkid) {
}
blk_id_ = blkid;
++pending_;
get_block_handle_checked(blkid, [=, manager = manager_, Self = actor_id(this)](td::Result<ConstBlockHandle> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
return;
}
td::actor::send_closure_later(
manager, &ValidatorManager::get_shard_state_from_db, R.move_as_ok(),
[=](td::Result<Ref<ShardState>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query,
res.move_as_error_prefix("cannot load state for "s + blkid.to_str() + " : "));
} else {
td::actor::send_closure_later(Self, &LiteQuery::got_block_state, blkid, res.move_as_ok());
}
});
});
td::actor::send_closure(manager_, &ValidatorManager::get_block_state_for_litequery, blkid,
[Self = actor_id(this), blkid](td::Result<Ref<ShardState>> res) {
if (res.is_error()) {
td::actor::send_closure(
Self, &LiteQuery::abort_query,
res.move_as_error_prefix("cannot load state for "s + blkid.to_str() + " : "));
} else {
td::actor::send_closure_later(Self, &LiteQuery::got_block_state, blkid, res.move_as_ok());
}
});
return true;
}
@ -699,22 +687,16 @@ bool LiteQuery::request_block_data(BlockIdExt blkid) {
}
blk_id_ = blkid;
++pending_;
get_block_handle_checked(blkid, [=, manager = manager_, Self = actor_id(this)](td::Result<ConstBlockHandle> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
return;
}
td::actor::send_closure_later(
manager, &ValidatorManager::get_block_data_from_db, R.move_as_ok(),
[=](td::Result<Ref<BlockData>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query,
res.move_as_error_prefix("cannot load block "s + blkid.to_str() + " : "));
} else {
td::actor::send_closure_later(Self, &LiteQuery::got_block_data, blkid, res.move_as_ok());
}
});
});
td::actor::send_closure(manager_, &ValidatorManager::get_block_data_for_litequery, blkid,
[Self = actor_id(this), blkid](td::Result<Ref<BlockData>> res) {
if (res.is_error()) {
td::actor::send_closure(
Self, &LiteQuery::abort_query,
res.move_as_error_prefix("cannot load block "s + blkid.to_str() + " : "));
} else {
td::actor::send_closure_later(Self, &LiteQuery::got_block_data, blkid, res.move_as_ok());
}
});
return true;
}
@ -1712,7 +1694,7 @@ void LiteQuery::continue_getTransactions(unsigned remaining, bool exact) {
LOG(DEBUG) << "sending get_block_by_lt_from_db() query to manager for " << acc_workchain_ << ":" << acc_addr_.to_hex()
<< " " << trans_lt_;
td::actor::send_closure_later(
manager_, &ValidatorManager::get_block_by_lt_from_db_for_litequery, ton::extract_addr_prefix(acc_workchain_, acc_addr_),
manager_, &ValidatorManager::get_block_by_lt_for_litequery, ton::extract_addr_prefix(acc_workchain_, acc_addr_),
trans_lt_, [Self = actor_id(this), remaining, manager = manager_](td::Result<ConstBlockHandle> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_getTransactions, res.move_as_error(), ton::BlockIdExt{});
@ -2088,11 +2070,13 @@ void LiteQuery::perform_lookupBlockWithProof(BlockId blkid, BlockIdExt mc_blkid,
});
if (mode & 2) {
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_lt_from_db_for_litequery, pfx, lt, std::move(P));
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_lt_for_litequery, pfx, lt, std::move(P));
} else if (mode & 4) {
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_unix_time_from_db_for_litequery, pfx, utime, std::move(P));
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_unix_time_for_litequery, pfx, utime,
std::move(P));
} else {
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_seqno_from_db_for_litequery, pfx, blkid.seqno, std::move(P));
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_seqno_for_litequery, pfx, blkid.seqno,
std::move(P));
}
}
@ -2368,13 +2352,13 @@ void LiteQuery::perform_lookupBlock(BlockId blkid, int mode, LogicalTime lt, Uni
ton::AccountIdPrefixFull pfx{blkid.workchain, blkid.shard};
if (mode & 2) {
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_lt_from_db_for_litequery, pfx, lt,
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_lt_for_litequery, pfx, lt,
std::move(P));
} else if (mode & 4) {
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_unix_time_from_db_for_litequery, pfx, utime,
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_unix_time_for_litequery, pfx, utime,
std::move(P));
} else {
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_seqno_from_db_for_litequery, pfx,
td::actor::send_closure_later(manager_, &ValidatorManager::get_block_by_seqno_for_litequery, pfx,
blkid.seqno, std::move(P));
}
}
@ -3133,7 +3117,7 @@ void LiteQuery::perform_getShardBlockProof(BlockIdExt blkid) {
}
AccountIdPrefixFull pfx{masterchainId, shardIdAll};
td::actor::send_closure_later(
manager, &ValidatorManager::get_block_by_seqno_from_db_for_litequery, pfx, handle->masterchain_ref_block(),
manager, &ValidatorManager::get_block_by_seqno_for_litequery, pfx, handle->masterchain_ref_block(),
[Self, manager](td::Result<ConstBlockHandle> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
@ -3237,5 +3221,92 @@ void LiteQuery::continue_getShardBlockProof(Ref<BlockData> cur_block,
});
}
void LiteQuery::perform_getOutMsgQueueSizes(td::optional<ShardIdFull> shard) {
LOG(INFO) << "started a getOutMsgQueueSizes" << (shard ? shard.value().to_str() : "") << " liteserver query";
td::actor::send_closure_later(
manager_, &ton::validator::ValidatorManager::get_last_liteserver_state_block,
[Self = actor_id(this), shard](td::Result<std::pair<Ref<MasterchainState>, BlockIdExt>> res) {
if (res.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, res.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::continue_getOutMsgQueueSizes, shard, res.ok().first);
}
});
}
void LiteQuery::continue_getOutMsgQueueSizes(td::optional<ShardIdFull> shard, Ref<MasterchainState> state) {
std::vector<BlockIdExt> blocks;
if (!shard || shard_intersects(shard.value(), state->get_shard())) {
blocks.push_back(state->get_block_id());
}
for (auto& x : state->get_shards()) {
if (!shard || shard_intersects(shard.value(), x->shard())) {
blocks.push_back(x->top_block_id());
}
}
auto res = std::make_shared<std::vector<tl_object_ptr<lite_api::liteServer_outMsgQueueSize>>>(blocks.size());
td::MultiPromise mp;
auto ig = mp.init_guard();
for (size_t i = 0; i < blocks.size(); ++i) {
td::actor::send_closure(manager_, &ValidatorManager::get_out_msg_queue_size, blocks[i],
[promise = ig.get_promise(), res, i, id = blocks[i]](td::Result<td::uint32> R) mutable {
TRY_RESULT_PROMISE(promise, value, std::move(R));
res->at(i) = create_tl_object<lite_api::liteServer_outMsgQueueSize>(
create_tl_lite_block_id(id), value);
promise.set_value(td::Unit());
});
}
ig.add_promise([Self = actor_id(this), res](td::Result<td::Unit> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
return;
}
td::actor::send_closure(Self, &LiteQuery::finish_query,
create_serialize_tl_object<lite_api::liteServer_outMsgQueueSizes>(
std::move(*res), Collator::get_skip_externals_queue_size()),
false);
});
}
void LiteQuery::perform_nonfinal_getCandidate(td::Bits256 source, BlockIdExt blkid, td::Bits256 collated_data_hash) {
LOG(INFO) << "started a nonfinal.getCandidate liteserver query";
td::actor::send_closure_later(
manager_, &ValidatorManager::get_block_candidate_for_litequery, PublicKey{pubkeys::Ed25519{source}}, blkid, collated_data_hash,
[Self = actor_id(this)](td::Result<BlockCandidate> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
} else {
BlockCandidate cand = R.move_as_ok();
td::actor::send_closure_later(
Self, &LiteQuery::finish_query,
create_serialize_tl_object<lite_api::liteServer_nonfinal_candidate>(
create_tl_object<lite_api::liteServer_nonfinal_candidateId>(
create_tl_lite_block_id(cand.id), cand.pubkey.as_bits256(), cand.collated_file_hash),
std::move(cand.data), std::move(cand.collated_data)),
false);
}
});
}
void LiteQuery::perform_nonfinal_getValidatorGroups(int mode, ShardIdFull shard) {
bool with_shard = mode & 1;
LOG(INFO) << "started a nonfinal.getValidatorGroups" << (with_shard ? shard.to_str() : "(all)")
<< " liteserver query";
td::optional<ShardIdFull> maybe_shard;
if (with_shard) {
maybe_shard = shard;
}
td::actor::send_closure(
manager_, &ValidatorManager::get_validator_groups_info_for_litequery, maybe_shard,
[Self = actor_id(this)](td::Result<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups>> R) {
if (R.is_error()) {
td::actor::send_closure(Self, &LiteQuery::abort_query, R.move_as_error());
} else {
td::actor::send_closure_later(Self, &LiteQuery::finish_query, serialize_tl_object(R.move_as_ok(), true),
false);
}
});
}
} // namespace validator
} // namespace ton

5
validator/impl/liteserver.hpp
View file

@ -168,6 +168,11 @@ class LiteQuery : public td::actor::Actor {
void perform_getShardBlockProof(BlockIdExt blkid);
void continue_getShardBlockProof(Ref<BlockData> cur_block,
std::vector<std::pair<BlockIdExt, td::BufferSlice>> result);
void perform_getOutMsgQueueSizes(td::optional<ShardIdFull> shard);
void continue_getOutMsgQueueSizes(td::optional<ShardIdFull> shard, Ref<MasterchainState> state);
void perform_nonfinal_getCandidate(td::Bits256 source, BlockIdExt blkid, td::Bits256 collated_data_hash);
void perform_nonfinal_getValidatorGroups(int mode, ShardIdFull shard);
void load_prevKeyBlock(ton::BlockIdExt blkid, td::Promise<std::pair<BlockIdExt, Ref<BlockQ>>>);
void continue_loadPrevKeyBlock(ton::BlockIdExt blkid, td::Result<std::pair<Ref<MasterchainState>, BlockIdExt>> res,

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

@ -29,6 +29,7 @@
#include "liteserver.h"
#include "crypto/vm/db/DynamicBagOfCellsDb.h"
#include "validator-session/validator-session-types.h"
#include "auto/tl/lite_api.h"
#include "impl/out-msg-queue-proof.hpp"
namespace ton {
@ -171,12 +172,19 @@ class ValidatorManager : public ValidatorManagerInterface {
virtual void log_validator_session_stats(BlockIdExt block_id, validatorsession::ValidatorSessionStats stats) = 0;
virtual void get_block_handle_for_litequery(BlockIdExt block_id, td::Promise<ConstBlockHandle> promise) = 0;
virtual void get_block_by_lt_from_db_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
td::Promise<ConstBlockHandle> promise) = 0;
virtual void get_block_by_unix_time_from_db_for_litequery(AccountIdPrefixFull account, UnixTime ts,
td::Promise<ConstBlockHandle> promise) = 0;
virtual void get_block_by_seqno_from_db_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
td::Promise<ConstBlockHandle> promise) = 0;
virtual void get_block_data_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<BlockData>> promise) = 0;
virtual void get_block_state_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<ShardState>> promise) = 0;
virtual void get_block_by_lt_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
td::Promise<ConstBlockHandle> promise) = 0;
virtual void get_block_by_unix_time_for_litequery(AccountIdPrefixFull account, UnixTime ts,
td::Promise<ConstBlockHandle> promise) = 0;
virtual void get_block_by_seqno_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
td::Promise<ConstBlockHandle> promise) = 0;
virtual void get_block_candidate_for_litequery(PublicKey source, BlockIdExt block_id, FileHash collated_data_hash,
td::Promise<BlockCandidate> promise) = 0;
virtual void get_validator_groups_info_for_litequery(
td::optional<ShardIdFull> shard,
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups>> promise) = 0;
virtual void add_lite_query_stats(int lite_query_id) {
}

21
validator/manager-disk.hpp
View file

@ -396,18 +396,33 @@ class ValidatorManagerImpl : public ValidatorManager {
void get_block_handle_for_litequery(BlockIdExt block_id, td::Promise<ConstBlockHandle> promise) override {
get_block_handle(block_id, false, promise.wrap([](BlockHandle &&handle) -> ConstBlockHandle { return handle; }));
}
void get_block_by_lt_from_db_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
void get_block_data_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<BlockData>> promise) override {
get_block_data_from_db_short(block_id, std::move(promise));
}
void get_block_state_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<ShardState>> promise) override {
get_shard_state_from_db_short(block_id, std::move(promise));
}
void get_block_by_lt_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
td::Promise<ConstBlockHandle> promise) override {
get_block_by_lt_from_db(account, lt, std::move(promise));
}
void get_block_by_unix_time_from_db_for_litequery(AccountIdPrefixFull account, UnixTime ts,
void get_block_by_unix_time_for_litequery(AccountIdPrefixFull account, UnixTime ts,
td::Promise<ConstBlockHandle> promise) override {
get_block_by_unix_time_from_db(account, ts, std::move(promise));
}
void get_block_by_seqno_from_db_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
void get_block_by_seqno_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
td::Promise<ConstBlockHandle> promise) override {
get_block_by_seqno_from_db(account, seqno, std::move(promise));
}
void get_block_candidate_for_litequery(PublicKey source, BlockIdExt block_id, FileHash collated_data_hash,
td::Promise<BlockCandidate> promise) override {
promise.set_result(td::Status::Error("not implemented"));
}
void get_validator_groups_info_for_litequery(
td::optional<ShardIdFull> shard,
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups>> promise) override {
promise.set_result(td::Status::Error("not implemented"));
}
void validated_new_block(BlockIdExt block_id) override {
}
void add_persistent_state_description(td::Ref<PersistentStateDescription> desc) override {

21
validator/manager-hardfork.hpp
View file

@ -458,18 +458,33 @@ class ValidatorManagerImpl : public ValidatorManager {
void get_block_handle_for_litequery(BlockIdExt block_id, td::Promise<ConstBlockHandle> promise) override {
get_block_handle(block_id, false, promise.wrap([](BlockHandle &&handle) -> ConstBlockHandle { return handle; }));
}
void get_block_by_lt_from_db_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
void get_block_data_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<BlockData>> promise) override {
get_block_data_from_db_short(block_id, std::move(promise));
}
void get_block_state_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<ShardState>> promise) override {
get_shard_state_from_db_short(block_id, std::move(promise));
}
void get_block_by_lt_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
td::Promise<ConstBlockHandle> promise) override {
get_block_by_lt_from_db(account, lt, std::move(promise));
}
void get_block_by_unix_time_from_db_for_litequery(AccountIdPrefixFull account, UnixTime ts,
void get_block_by_unix_time_for_litequery(AccountIdPrefixFull account, UnixTime ts,
td::Promise<ConstBlockHandle> promise) override {
get_block_by_unix_time_from_db(account, ts, std::move(promise));
}
void get_block_by_seqno_from_db_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
void get_block_by_seqno_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
td::Promise<ConstBlockHandle> promise) override {
get_block_by_seqno_from_db(account, seqno, std::move(promise));
}
void get_block_candidate_for_litequery(PublicKey source, BlockIdExt block_id, FileHash collated_data_hash,
td::Promise<BlockCandidate> promise) override {
promise.set_result(td::Status::Error("not implemented"));
}
void get_validator_groups_info_for_litequery(
td::optional<ShardIdFull> shard,
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups>> promise) override {
promise.set_result(td::Status::Error("not implemented"));
}
void validated_new_block(BlockIdExt block_id) override {
}
void add_persistent_state_description(td::Ref<PersistentStateDescription> desc) override {

198
validator/manager.cpp
View file

@ -227,13 +227,13 @@ void ValidatorManagerImpl::sync_complete(td::Promise<td::Unit> promise) {
VLOG(VALIDATOR_WARNING) << "completed sync. Validating " << validator_groups_.size() << " groups";
for (auto &v : validator_groups_) {
if (!v.second.empty()) {
td::actor::send_closure(v.second, &ValidatorGroup::create_session);
if (!v.second.actor.empty()) {
td::actor::send_closure(v.second.actor, &ValidatorGroup::create_session);
}
}
for (auto &v : next_validator_groups_) {
if (!v.second.empty()) {
td::actor::send_closure(v.second, &ValidatorGroup::create_session);
if (!v.second.actor.empty()) {
td::actor::send_closure(v.second.actor, &ValidatorGroup::create_session);
}
}
}
@ -1282,6 +1282,10 @@ void ValidatorManagerImpl::set_next_block(BlockIdExt block_id, BlockIdExt next,
}
void ValidatorManagerImpl::set_block_candidate(BlockIdExt id, BlockCandidate candidate, td::Promise<td::Unit> promise) {
if (!candidates_buffer_.empty()) {
td::actor::send_closure(candidates_buffer_, &CandidatesBuffer::add_new_candidate, id,
PublicKey{pubkeys::Ed25519{candidate.pubkey.as_bits256()}}, candidate.collated_file_hash);
}
td::actor::send_closure(db_, &Db::store_block_candidate, std::move(candidate), std::move(promise));
}
@ -1696,9 +1700,12 @@ void ValidatorManagerImpl::started(ValidatorManagerInitResult R) {
td::actor::send_closure(SelfId, &ValidatorManagerImpl::read_gc_list, R.move_as_ok());
}
});
td::actor::send_closure(db_, &Db::get_destroyed_validator_sessions, std::move(P));
if (opts_->nonfinal_ls_queries_enabled()) {
candidates_buffer_ = td::actor::create_actor<CandidatesBuffer>("candidates-buffer", actor_id(this));
}
auto Q = td::PromiseCreator::lambda(
[SelfId = actor_id(this)](td::Result<std::vector<td::Ref<PersistentStateDescription>>> R) {
if (R.is_error()) {
@ -2025,8 +2032,8 @@ void ValidatorManagerImpl::update_shards() {
VLOG(VALIDATOR_DEBUG) << "total shards=" << new_shards.size() << " config shards=" << exp_vec.size();
std::map<ValidatorSessionId, td::actor::ActorOwn<ValidatorGroup>> new_validator_groups_;
std::map<ValidatorSessionId, td::actor::ActorOwn<ValidatorGroup>> new_next_validator_groups_;
std::map<ValidatorSessionId, ValidatorGroupEntry> new_validator_groups_;
std::map<ValidatorSessionId, ValidatorGroupEntry> new_next_validator_groups_;
bool force_recover = false;
{
@ -2058,8 +2065,8 @@ void ValidatorManagerImpl::update_shards() {
} else {
auto it2 = next_validator_groups_.find(legacy_val_group_id);
if (it2 != next_validator_groups_.end()) {
if (!it2->second.empty()) {
td::actor::send_closure(it2->second, &ValidatorGroup::start, prev, last_masterchain_block_id_);
if (!it2->second.actor.empty()) {
td::actor::send_closure(it2->second.actor, &ValidatorGroup::start, prev, last_masterchain_block_id_);
}
new_validator_groups_.emplace(val_group_id, std::move(it2->second));
} else {
@ -2067,7 +2074,7 @@ void ValidatorManagerImpl::update_shards() {
if (!G.empty()) {
td::actor::send_closure(G, &ValidatorGroup::start, prev, last_masterchain_block_id_);
}
new_validator_groups_.emplace(val_group_id, std::move(G));
new_validator_groups_.emplace(val_group_id, ValidatorGroupEntry{std::move(G), shard});
}
}
}
@ -2113,8 +2120,8 @@ void ValidatorManagerImpl::update_shards() {
} else {
auto it2 = next_validator_groups_.find(val_group_id);
if (it2 != next_validator_groups_.end()) {
if (!it2->second.empty()) {
td::actor::send_closure(it2->second, &ValidatorGroup::start, prev, last_masterchain_block_id_);
if (!it2->second.actor.empty()) {
td::actor::send_closure(it2->second.actor, &ValidatorGroup::start, prev, last_masterchain_block_id_);
}
new_validator_groups_.emplace(val_group_id, std::move(it2->second));
} else {
@ -2122,7 +2129,7 @@ void ValidatorManagerImpl::update_shards() {
if (!G.empty()) {
td::actor::send_closure(G, &ValidatorGroup::start, prev, last_masterchain_block_id_);
}
new_validator_groups_.emplace(val_group_id, std::move(G));
new_validator_groups_.emplace(val_group_id, ValidatorGroupEntry{std::move(G), shard});
}
}
}
@ -2142,23 +2149,24 @@ void ValidatorManagerImpl::update_shards() {
//CHECK(!it->second.empty());
new_next_validator_groups_.emplace(val_group_id, std::move(it->second));
} else {
new_next_validator_groups_.emplace(val_group_id,
create_validator_group(val_group_id, shard, val_set, opts, started_));
new_next_validator_groups_.emplace(
val_group_id,
ValidatorGroupEntry{create_validator_group(val_group_id, shard, val_set, opts, started_), shard});
}
}
}
std::vector<td::actor::ActorId<ValidatorGroup>> gc;
for (auto &v : validator_groups_) {
if (!v.second.empty()) {
if (!v.second.actor.empty()) {
gc_list_.push_back(v.first);
gc.push_back(v.second.release());
gc.push_back(v.second.actor.release());
}
}
for (auto &v : next_validator_groups_) {
if (!v.second.empty()) {
if (!v.second.actor.empty()) {
gc_list_.push_back(v.first);
gc.push_back(v.second.release());
gc.push_back(v.second.actor.release());
}
}
@ -2903,19 +2911,72 @@ void ValidatorManagerImpl::log_validator_session_stats(BlockIdExt block_id,
}
void ValidatorManagerImpl::get_block_handle_for_litequery(BlockIdExt block_id, td::Promise<ConstBlockHandle> promise) {
get_block_handle(
block_id, false,
[SelfId = actor_id(this), block_id, promise = std::move(promise)](td::Result<BlockHandle> R) mutable {
if (R.is_ok() && R.ok()->is_applied()) {
promise.set_value(R.move_as_ok());
} else {
td::actor::send_closure(SelfId, &ValidatorManagerImpl::process_block_handle_for_litequery_error, block_id,
std::move(R), std::move(promise));
}
});
get_block_handle(block_id, false,
[SelfId = actor_id(this), block_id, promise = std::move(promise),
allow_not_applied = opts_->nonfinal_ls_queries_enabled()](td::Result<BlockHandle> R) mutable {
if (R.is_ok() && (allow_not_applied || R.ok()->is_applied())) {
promise.set_value(R.move_as_ok());
} else {
td::actor::send_closure(SelfId, &ValidatorManagerImpl::process_block_handle_for_litequery_error,
block_id, std::move(R), std::move(promise));
}
});
}
void ValidatorManagerImpl::get_block_by_lt_from_db_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
void ValidatorManagerImpl::get_block_data_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<BlockData>> promise) {
if (candidates_buffer_.empty()) {
get_block_handle_for_litequery(
block_id, [manager = actor_id(this), promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
TRY_RESULT_PROMISE(promise, handle, std::move(R));
td::actor::send_closure_later(manager, &ValidatorManager::get_block_data_from_db, std::move(handle),
std::move(promise));
});
} else {
td::actor::send_closure(
candidates_buffer_, &CandidatesBuffer::get_block_data, block_id,
[manager = actor_id(this), promise = std::move(promise), block_id](td::Result<td::Ref<BlockData>> R) mutable {
if (R.is_ok()) {
promise.set_result(R.move_as_ok());
return;
}
td::actor::send_closure(manager, &ValidatorManagerImpl::get_block_handle_for_litequery, block_id,
[manager, promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
TRY_RESULT_PROMISE(promise, handle, std::move(R));
td::actor::send_closure_later(manager, &ValidatorManager::get_block_data_from_db,
std::move(handle), std::move(promise));
});
});
}
}
void ValidatorManagerImpl::get_block_state_for_litequery(BlockIdExt block_id,
td::Promise<td::Ref<ShardState>> promise) {
if (candidates_buffer_.empty()) {
get_block_handle_for_litequery(
block_id, [manager = actor_id(this), promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
TRY_RESULT_PROMISE(promise, handle, std::move(R));
td::actor::send_closure_later(manager, &ValidatorManager::get_shard_state_from_db, std::move(handle),
std::move(promise));
});
} else {
td::actor::send_closure(
candidates_buffer_, &CandidatesBuffer::get_block_state, block_id,
[manager = actor_id(this), promise = std::move(promise), block_id](td::Result<td::Ref<ShardState>> R) mutable {
if (R.is_ok()) {
promise.set_result(R.move_as_ok());
return;
}
td::actor::send_closure(manager, &ValidatorManagerImpl::get_block_handle_for_litequery,
block_id, [manager, promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
TRY_RESULT_PROMISE(promise, handle, std::move(R));
td::actor::send_closure_later(manager, &ValidatorManager::get_shard_state_from_db, std::move(handle),
std::move(promise));
});
});
}
}
void ValidatorManagerImpl::get_block_by_lt_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
td::Promise<ConstBlockHandle> promise) {
get_block_by_lt_from_db(
account, lt, [=, SelfId = actor_id(this), promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
@ -2928,7 +2989,7 @@ void ValidatorManagerImpl::get_block_by_lt_from_db_for_litequery(AccountIdPrefix
});
}
void ValidatorManagerImpl::get_block_by_unix_time_from_db_for_litequery(AccountIdPrefixFull account, UnixTime ts,
void ValidatorManagerImpl::get_block_by_unix_time_for_litequery(AccountIdPrefixFull account, UnixTime ts,
td::Promise<ConstBlockHandle> promise) {
get_block_by_unix_time_from_db(
account, ts, [=, SelfId = actor_id(this), promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
@ -2941,7 +3002,7 @@ void ValidatorManagerImpl::get_block_by_unix_time_from_db_for_litequery(AccountI
});
}
void ValidatorManagerImpl::get_block_by_seqno_from_db_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
void ValidatorManagerImpl::get_block_by_seqno_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
td::Promise<ConstBlockHandle> promise) {
get_block_by_seqno_from_db(
account, seqno,
@ -3039,11 +3100,82 @@ void ValidatorManagerImpl::process_lookup_block_for_litequery_error(AccountIdPre
promise.set_error(std::move(err));
}
void ValidatorManagerImpl::get_block_candidate_for_litequery(PublicKey source, BlockIdExt block_id,
FileHash collated_data_hash,
td::Promise<BlockCandidate> promise) {
if (!opts_->nonfinal_ls_queries_enabled()) {
promise.set_error(td::Status::Error("query is not allowed"));
return;
}
get_block_candidate_from_db(source, block_id, collated_data_hash, std::move(promise));
}
void ValidatorManagerImpl::get_validator_groups_info_for_litequery(
td::optional<ShardIdFull> shard,
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups>> promise) {
if (!opts_->nonfinal_ls_queries_enabled()) {
promise.set_error(td::Status::Error("query is not allowed"));
return;
}
class Actor : public td::actor::Actor {
public:
explicit Actor(std::vector<td::actor::ActorId<ValidatorGroup>> groups,
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups>> promise)
: groups_(std::move(groups)), promise_(std::move(promise)) {
}
void start_up() override {
pending_ = groups_.size();
if (pending_ == 0) {
promise_.set_result(std::move(result_));
stop();
return;
}
for (auto &x : groups_) {
td::actor::send_closure(
x, &ValidatorGroup::get_validator_group_info_for_litequery,
[SelfId = actor_id(this)](td::Result<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroupInfo>> R) {
td::actor::send_closure(SelfId, &Actor::on_result, R.is_ok() ? R.move_as_ok() : nullptr);
});
}
}
void on_result(tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroupInfo> r) {
if (r) {
result_->groups_.push_back(std::move(r));
}
--pending_;
if (pending_ == 0) {
promise_.set_result(std::move(result_));
stop();
}
}
private:
std::vector<td::actor::ActorId<ValidatorGroup>> groups_;
size_t pending_;
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups>> promise_;
tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups> result_ =
create_tl_object<lite_api::liteServer_nonfinal_validatorGroups>();
};
std::vector<td::actor::ActorId<ValidatorGroup>> groups;
for (auto &x : validator_groups_) {
if (x.second.actor.empty()) {
continue;
}
if (shard && shard.value() != x.second.shard) {
continue;
}
groups.push_back(x.second.actor.get());
}
td::actor::create_actor<Actor>("get-validator-groups-info", std::move(groups), std::move(promise)).release();
}
void ValidatorManagerImpl::get_validator_sessions_info(
td::Promise<tl_object_ptr<ton_api::engine_validator_validatorSessionsInfo>> promise) {
std::vector<td::actor::ActorId<ValidatorGroup>> groups;
for (const auto& g : validator_groups_) {
groups.push_back(g.second.get());
groups.push_back(g.second.actor.get());
}
struct IntermediateData {
std::vector<td::actor::ActorId<ValidatorGroup>> groups;

24
validator/manager.hpp
View file

@ -29,6 +29,7 @@
#include "rldp/rldp.h"
#include "token-manager.h"
#include "queue-size-counter.hpp"
#include "impl/candidates-buffer.hpp"
#include "collator-node.hpp"
#include <map>
@ -252,8 +253,12 @@ class ValidatorManagerImpl : public ValidatorManager {
td::Ref<ValidatorSet> validator_set,
validatorsession::ValidatorSessionOptions opts,
bool create_catchain);
std::map<ValidatorSessionId, td::actor::ActorOwn<ValidatorGroup>> validator_groups_;
std::map<ValidatorSessionId, td::actor::ActorOwn<ValidatorGroup>> next_validator_groups_;
struct ValidatorGroupEntry {
td::actor::ActorOwn<ValidatorGroup> actor;
ShardIdFull shard;
};
std::map<ValidatorSessionId, ValidatorGroupEntry> validator_groups_;
std::map<ValidatorSessionId, ValidatorGroupEntry> next_validator_groups_;
std::set<ValidatorSessionId> check_gc_list_;
std::vector<ValidatorSessionId> gc_list_;
@ -607,17 +612,24 @@ class ValidatorManagerImpl : public ValidatorManager {
}
void get_block_handle_for_litequery(BlockIdExt block_id, td::Promise<ConstBlockHandle> promise) override;
void get_block_by_lt_from_db_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
void get_block_data_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<BlockData>> promise) override;
void get_block_state_for_litequery(BlockIdExt block_id, td::Promise<td::Ref<ShardState>> promise) override;
void get_block_by_lt_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
td::Promise<ConstBlockHandle> promise) override;
void get_block_by_unix_time_from_db_for_litequery(AccountIdPrefixFull account, UnixTime ts,
void get_block_by_unix_time_for_litequery(AccountIdPrefixFull account, UnixTime ts,
td::Promise<ConstBlockHandle> promise) override;
void get_block_by_seqno_from_db_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
void get_block_by_seqno_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
td::Promise<ConstBlockHandle> promise) override;
void process_block_handle_for_litequery_error(BlockIdExt block_id, td::Result<BlockHandle> r_handle,
td::Promise<ConstBlockHandle> promise);
void process_lookup_block_for_litequery_error(AccountIdPrefixFull account, int type, td::uint64 value,
td::Result<ConstBlockHandle> r_handle,
td::Promise<ConstBlockHandle> promise);
void get_block_candidate_for_litequery(PublicKey source, BlockIdExt block_id, FileHash collated_data_hash,
td::Promise<BlockCandidate> promise) override;
void get_validator_groups_info_for_litequery(
td::optional<ShardIdFull> shard,
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroups>> promise) override;
void add_lite_query_stats(int lite_query_id) override {
++ls_stats_[lite_query_id];
@ -702,6 +714,8 @@ class ValidatorManagerImpl : public ValidatorManager {
std::map<int, td::uint32> ls_stats_; // lite_api ID -> count, 0 for unknown
td::uint32 ls_stats_check_ext_messages_{0};
td::actor::ActorOwn<CandidatesBuffer> candidates_buffer_;
struct Collator {
td::actor::ActorOwn<CollatorNode> actor;
std::set<ShardIdFull> shards;

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

@ -23,6 +23,7 @@
#include "td/utils/overloaded.h"
#include "ton/ton-io.hpp"
#include "validator/full-node.h"
#include "full-node-serializer.hpp"
namespace ton {
@ -219,52 +220,54 @@ void DownloadBlockNew::got_data(td::BufferSlice data) {
}
auto f = F.move_as_ok();
if (f->get_id() == ton_api::tonNode_dataFullEmpty::ID) {
abort_query(td::Status::Error(ErrorCode::notready, "node doesn't have this block"));
return;
}
BlockIdExt id;
td::BufferSlice proof, block_data;
bool is_link;
td::Status S = deserialize_block_full(*f, id, proof, block_data, is_link, overlay::Overlays::max_fec_broadcast_size());
if (S.is_error()) {
abort_query(S.move_as_error_prefix("cannot deserialize block: "));
return;
}
ton_api::downcast_call(
*f.get(),
td::overloaded(
[&](ton_api::tonNode_dataFullEmpty &x) {
abort_query(td::Status::Error(ErrorCode::notready, "node doesn't have this block"));
},
[&, self = this](ton_api::tonNode_dataFull &x) {
if (!allow_partial_proof_ && x.is_link_) {
abort_query(td::Status::Error(ErrorCode::notready, "node doesn't have proof for this block"));
return;
}
auto id = create_block_id(x.id_);
if (block_id_.is_valid() && id != block_id_) {
abort_query(td::Status::Error(ErrorCode::notready, "received data for wrong block"));
return;
}
block_.id = id;
block_.data = std::move(x.block_);
if (td::sha256_bits256(block_.data.as_slice()) != id.file_hash) {
abort_query(td::Status::Error(ErrorCode::notready, "received data with bad hash"));
return;
}
if (!allow_partial_proof_ && is_link) {
abort_query(td::Status::Error(ErrorCode::notready, "node doesn't have proof for this block"));
return;
}
if (block_id_.is_valid() && id != block_id_) {
abort_query(td::Status::Error(ErrorCode::notready, "received data for wrong block"));
return;
}
block_.id = id;
block_.data = std::move(block_data);
if (td::sha256_bits256(block_.data.as_slice()) != id.file_hash) {
abort_query(td::Status::Error(ErrorCode::notready, "received data with bad hash"));
return;
}
auto P = td::PromiseCreator::lambda([SelfId = actor_id(self)](td::Result<td::Unit> R) {
if (R.is_error()) {
td::actor::send_closure(SelfId, &DownloadBlockNew::abort_query,
R.move_as_error_prefix("received bad proof: "));
} else {
td::actor::send_closure(SelfId, &DownloadBlockNew::checked_block_proof);
}
});
if (block_id_.is_valid()) {
if (x.is_link_) {
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::validate_block_proof_link,
block_id_, std::move(x.proof_), std::move(P));
} else {
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::validate_block_proof, block_id_,
std::move(x.proof_), std::move(P));
}
} else {
CHECK(!x.is_link_);
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::validate_block_is_next_proof,
prev_id_, id, std::move(x.proof_), std::move(P));
}
}));
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::Unit> R) {
if (R.is_error()) {
td::actor::send_closure(SelfId, &DownloadBlockNew::abort_query, R.move_as_error_prefix("received bad proof: "));
} else {
td::actor::send_closure(SelfId, &DownloadBlockNew::checked_block_proof);
}
});
if (block_id_.is_valid()) {
if (is_link) {
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::validate_block_proof_link, block_id_,
std::move(proof), std::move(P));
} else {
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::validate_block_proof, block_id_,
std::move(proof), std::move(P));
}
} else {
CHECK(!is_link);
td::actor::send_closure(validator_manager_, &ValidatorManagerInterface::validate_block_is_next_proof, prev_id_, id,
std::move(proof), std::move(P));
}
}
void DownloadBlockNew::got_data_from_db(td::BufferSlice data) {

56
validator/validator-group.cpp
View file

@ -21,6 +21,7 @@
#include "ton/ton-io.hpp"
#include "td/utils/overloaded.h"
#include "common/delay.h"
#include "ton/lite-tl.hpp"
#include "ton/ton-tl.hpp"
#include "td/utils/Random.h"
@ -70,7 +71,9 @@ void ValidatorGroup::generated_block_candidate(std::shared_ptr<CachedCollatedBlo
cached_collated_block_ = nullptr;
}
} else {
cache->result = R.move_as_ok();
auto candidate = R.move_as_ok();
add_available_block_candidate(candidate.pubkey.as_bits256(), candidate.id, candidate.collated_file_hash);
cache->result = std::move(candidate);
for (auto &p : cache->promises) {
p.set_value(cache->result.value().clone());
}
@ -117,6 +120,8 @@ void ValidatorGroup::validate_block_candidate(td::uint32 round_id, BlockCandidat
[&](UnixTime ts) {
td::actor::send_closure(SelfId, &ValidatorGroup::update_approve_cache, block_to_cache_key(block),
ts);
td::actor::send_closure(SelfId, &ValidatorGroup::add_available_block_candidate, block.pubkey.as_bits256(),
block.id, block.collated_file_hash);
promise.set_result(ts);
},
[&](CandidateReject reject) {
@ -213,6 +218,10 @@ void ValidatorGroup::get_approved_candidate(PublicKey source, RootHash root_hash
std::move(promise));
}
BlockIdExt ValidatorGroup::create_next_block_id(RootHash root_hash, FileHash file_hash) const {
return BlockIdExt{create_next_block_id_simple(), root_hash, file_hash};
}
BlockId ValidatorGroup::create_next_block_id_simple() const {
BlockSeqno seqno = 0;
for (auto &p : prev_block_ids_) {
@ -223,10 +232,6 @@ BlockId ValidatorGroup::create_next_block_id_simple() const {
return BlockId{shard_.workchain, shard_.shard, seqno + 1};
}
BlockIdExt ValidatorGroup::create_next_block_id(RootHash root_hash, FileHash file_hash) const {
return BlockIdExt{create_next_block_id_simple(), root_hash, file_hash};
}
std::unique_ptr<validatorsession::ValidatorSession::Callback> ValidatorGroup::make_validator_session_callback() {
class Callback : public validatorsession::ValidatorSession::Callback {
public:
@ -382,6 +387,47 @@ void ValidatorGroup::destroy() {
stop();
}
void ValidatorGroup::get_validator_group_info_for_litequery(
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroupInfo>> promise) {
if (session_.empty()) {
promise.set_error(td::Status::Error(ErrorCode::notready, "not started"));
return;
}
td::actor::send_closure(
session_, &validatorsession::ValidatorSession::get_validator_group_info_for_litequery, last_known_round_id_,
[SelfId = actor_id(this), promise = std::move(promise), round = last_known_round_id_](
td::Result<std::vector<tl_object_ptr<lite_api::liteServer_nonfinal_candidateInfo>>> R) mutable {
TRY_RESULT_PROMISE(promise, result, std::move(R));
td::actor::send_closure(SelfId, &ValidatorGroup::get_validator_group_info_for_litequery_cont, round,
std::move(result), std::move(promise));
});
}
void ValidatorGroup::get_validator_group_info_for_litequery_cont(
td::uint32 expected_round, std::vector<tl_object_ptr<lite_api::liteServer_nonfinal_candidateInfo>> candidates,
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroupInfo>> promise) {
if (expected_round != last_known_round_id_) {
candidates.clear();
}
BlockId next_block_id = create_next_block_id_simple();
for (auto &candidate : candidates) {
BlockIdExt id{next_block_id, candidate->id_->block_id_->root_hash_, candidate->id_->block_id_->file_hash_};
candidate->id_->block_id_ = create_tl_lite_block_id(id);
candidate->available_ =
available_block_candidates_.count({candidate->id_->creator_, id, candidate->id_->collated_data_hash_});
}
auto result = create_tl_object<lite_api::liteServer_nonfinal_validatorGroupInfo>();
result->next_block_id_ = create_tl_lite_block_id_simple(next_block_id);
for (const BlockIdExt& prev : prev_block_ids_) {
result->prev_.push_back(create_tl_lite_block_id(prev));
}
result->cc_seqno_ = validator_set_->get_catchain_seqno();
result->candidates_ = std::move(candidates);
promise.set_result(std::move(result));
}
void ValidatorGroup::get_session_info(
td::Promise<tl_object_ptr<ton_api::engine_validator_validatorSessionInfo>> promise) {
if (session_.empty() || !started_) {

16
validator/validator-group.hpp
View file

@ -46,8 +46,8 @@ class ValidatorGroup : public td::actor::Actor {
bool send_broadcast, td::Promise<td::Unit> promise, bool is_retry = false);
void get_approved_candidate(PublicKey source, RootHash root_hash, FileHash file_hash,
FileHash collated_data_file_hash, td::Promise<BlockCandidate> promise);
BlockId create_next_block_id_simple() const;
BlockIdExt create_next_block_id(RootHash root_hash, FileHash file_hash) const;
BlockId create_next_block_id_simple() const;
void start(std::vector<BlockIdExt> prev, BlockIdExt min_masterchain_block_id);
void create_session();
@ -59,6 +59,9 @@ class ValidatorGroup : public td::actor::Actor {
}
}
void get_validator_group_info_for_litequery(
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroupInfo>> promise);
void get_session_info(td::Promise<tl_object_ptr<ton_api::engine_validator_validatorSessionInfo>> promise);
ValidatorGroup(ShardIdFull shard, PublicKeyHash local_id, ValidatorSessionId session_id,
@ -147,6 +150,17 @@ class ValidatorGroup : public td::actor::Actor {
static CacheKey block_to_cache_key(const BlockCandidate& block) {
return std::make_tuple(block.pubkey.as_bits256(), block.id, sha256_bits256(block.data), block.collated_file_hash);
}
void get_validator_group_info_for_litequery_cont(
td::uint32 expected_round,
std::vector<tl_object_ptr<lite_api::liteServer_nonfinal_candidateInfo>> candidates,
td::Promise<tl_object_ptr<lite_api::liteServer_nonfinal_validatorGroupInfo>> promise);
std::set<std::tuple<td::Bits256, BlockIdExt, FileHash>> available_block_candidates_; // source, id, collated hash
void add_available_block_candidate(td::Bits256 source, BlockIdExt id, FileHash collated_data_hash) {
available_block_candidates_.emplace(source, id, collated_data_hash);
}
};
} // namespace validator

7
validator/validator-options.hpp
View file

@ -124,6 +124,9 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
bool get_disable_rocksdb_stats() const override {
return disable_rocksdb_stats_;
}
bool nonfinal_ls_queries_enabled() const override {
return nonfinal_ls_queries_enabled_;
}
ValidatorMode validator_mode() const override {
return validator_mode_;
}
@ -192,6 +195,9 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
void set_disable_rocksdb_stats(bool value) override {
disable_rocksdb_stats_ = value;
}
void set_nonfinal_ls_queries_enabled(bool value) override {
nonfinal_ls_queries_enabled_ = value;
}
void set_validator_mode(ValidatorMode value) override {
validator_mode_ = value;
}
@ -239,6 +245,7 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
size_t max_open_archive_files_ = 0;
double archive_preload_period_ = 0.0;
bool disable_rocksdb_stats_;
bool nonfinal_ls_queries_enabled_ = false;
ValidatorMode validator_mode_ = validator_normal;
};

2
validator/validator.h
View file

@ -85,6 +85,7 @@ struct ValidatorManagerOptions : public td::CntObject {
virtual size_t get_max_open_archive_files() const = 0;
virtual double get_archive_preload_period() const = 0;
virtual bool get_disable_rocksdb_stats() const = 0;
virtual bool nonfinal_ls_queries_enabled() const = 0;
virtual ValidatorMode validator_mode() const = 0;
virtual void set_zero_block_id(BlockIdExt block_id) = 0;
@ -108,6 +109,7 @@ struct ValidatorManagerOptions : public td::CntObject {
virtual void set_max_open_archive_files(size_t value) = 0;
virtual void set_archive_preload_period(double value) = 0;
virtual void set_disable_rocksdb_stats(bool value) = 0;
virtual void set_nonfinal_ls_queries_enabled(bool value) = 0;
virtual void set_validator_mode(ValidatorMode value) = 0;
static td::Ref<ValidatorManagerOptions> create(