external/ton
1
0
Fork 0
mirror of https://github.com/ton-blockchain/ton synced 2025-02-12 19:22:37 +00:00
Code Issues Releases Wiki Activity
ton/overlay/overlay-manager.cpp
EmelyanenkoK 0434eadc1f
Add custom overlays for external messages (#949) 
* Private overlay for external messages

* Improve ext msg overlays

* Manage from validator console
* Bypass out queue size limit for high-priority messages
* Shuffle messages in get_external_messages

* Cleanup mempool when creating validator group

* Improve private overlays for externals

1. Allow using validator adnl ids in addition to fullnode ids
2. Set priority per sender, not per overlay
3. Require the same overlay name for all nodes
4. Enable lz4 in private block overlay

* Fix typo, add debug logs

* Enable lz4 in private block overlay by config

Change proto_version for lz4 in catchain overlays to 4

* Add logs for broadcasts in fullnode

---------

Co-authored-by: SpyCheese <mikle98@yandex.ru>
2024-04-01 16:44:08 +03:00

459 lines
20 KiB
C++
Raw Permalink Blame History

/*
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/>.
Copyright 2017-2020 Telegram Systems LLP
*/
#include "overlay-manager.h"
#include "auto/tl/ton_api.h"
#include "overlay.h"
#include "adnl/utils.hpp"
#include "td/actor/actor.h"
#include "td/actor/common.h"
#include "td/utils/Random.h"
#include "td/db/RocksDb.h"
#include "td/utils/Status.h"
#include "td/utils/overloaded.h"
#include "keys/encryptor.h"
#include "td/utils/port/Poll.h"
#include <vector>
namespace ton {
namespace overlay {
void OverlayManager::update_dht_node(td::actor::ActorId<dht::Dht> dht) {
dht_node_ = dht;
for (auto &X : overlays_) {
for (auto &Y : X.second) {
td::actor::send_closure(Y.second, &Overlay::update_dht_node, dht);
}
}
}
void OverlayManager::register_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id,
td::actor::ActorOwn<Overlay> overlay) {
auto it = overlays_.find(local_id);
VLOG(OVERLAY_INFO) << this << ": registering overlay " << overlay_id << "@" << local_id;
if (it == overlays_.end()) {
td::actor::send_closure(adnl_, &adnl::Adnl::subscribe, local_id,
adnl::Adnl::int_to_bytestring(ton_api::overlay_message::ID),
std::make_unique<AdnlCallback>(actor_id(this)));
td::actor::send_closure(adnl_, &adnl::Adnl::subscribe, local_id,
adnl::Adnl::int_to_bytestring(ton_api::overlay_query::ID),
std::make_unique<AdnlCallback>(actor_id(this)));
}
overlays_[local_id][overlay_id] = std::move(overlay);
auto P = td::PromiseCreator::lambda([id = overlays_[local_id][overlay_id].get()](td::Result<DbType::GetResult> R) {
R.ensure();
auto value = R.move_as_ok();
if (value.status == td::KeyValue::GetStatus::Ok) {
auto F = fetch_tl_object<ton_api::overlay_db_nodes>(std::move(value.value), true);
F.ensure();
auto nodes = std::move(F.move_as_ok()->nodes_);
td::actor::send_closure(id, &Overlay::receive_nodes_from_db, std::move(nodes));
}
});
auto key = create_hash_tl_object<ton_api::overlay_db_key_nodes>(local_id.bits256_value(), overlay_id.bits256_value());
db_.get(key, std::move(P));
}
void OverlayManager::delete_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id) {
auto it = overlays_.find(local_id);
if (it != overlays_.end()) {
it->second.erase(overlay_id);
if (it->second.size() == 0) {
td::actor::send_closure(adnl_, &adnl::Adnl::unsubscribe, local_id,
adnl::Adnl::int_to_bytestring(ton_api::overlay_message::ID));
td::actor::send_closure(adnl_, &adnl::Adnl::unsubscribe, local_id,
adnl::Adnl::int_to_bytestring(ton_api::overlay_query::ID));
overlays_.erase(it);
}
}
}
void OverlayManager::create_public_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
std::unique_ptr<Callback> callback, OverlayPrivacyRules rules,
td::string scope) {
create_public_overlay_ex(local_id, std::move(overlay_id), std::move(callback), std::move(rules), std::move(scope),
{});
}
void OverlayManager::create_public_overlay_ex(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
std::unique_ptr<Callback> callback, OverlayPrivacyRules rules,
td::string scope, OverlayOptions opts) {
CHECK(!dht_node_.empty());
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, std::move(opts)));
}
void OverlayManager::create_private_overlay(adnl::AdnlNodeIdShort local_id, OverlayIdFull overlay_id,
std::vector<adnl::AdnlNodeIdShort> nodes,
std::unique_ptr<Callback> callback, OverlayPrivacyRules rules,
std::string scope) {
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(nodes), std::move(callback), std::move(rules), std::move(scope)));
}
void OverlayManager::receive_message(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::BufferSlice data) {
auto R = fetch_tl_prefix<ton_api::overlay_message>(data, true);
if (R.is_error()) {
VLOG(OVERLAY_WARNING) << this << ": can not parse overlay message: " << R.move_as_error();
return;
}
auto M = R.move_as_ok();
auto it = overlays_.find(dst);
if (it == overlays_.end()) {
VLOG(OVERLAY_NOTICE) << this << ": message to unknown overlay " << M->overlay_ << "@" << dst;
return;
}
auto it2 = it->second.find(OverlayIdShort{M->overlay_});
if (it2 == it->second.end()) {
VLOG(OVERLAY_NOTICE) << this << ": message to localid is not in overlay " << M->overlay_ << "@" << dst;
return;
}
td::actor::send_closure(it2->second, &Overlay::update_throughput_in_ctr, src, (td::uint32)data.size(), false);
td::actor::send_closure(it2->second, &Overlay::receive_message, src, std::move(data));
}
void OverlayManager::receive_query(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, td::BufferSlice data,
td::Promise<td::BufferSlice> promise) {
auto R = fetch_tl_prefix<ton_api::overlay_query>(data, true);
if (R.is_error()) {
VLOG(OVERLAY_WARNING) << this << ": can not parse overlay query [" << src << "->" << dst
<< "]: " << R.move_as_error();
promise.set_error(td::Status::Error(ErrorCode::protoviolation, "bad overlay query header"));
return;
}
auto M = R.move_as_ok();
auto it = overlays_.find(dst);
if (it == overlays_.end()) {
VLOG(OVERLAY_NOTICE) << this << ": query to unknown overlay " << M->overlay_ << "@" << dst << " from " << src;
promise.set_error(td::Status::Error(ErrorCode::protoviolation, PSTRING() << "bad local_id " << dst));
return;
}
auto it2 = it->second.find(OverlayIdShort{M->overlay_});
if (it2 == it->second.end()) {
VLOG(OVERLAY_NOTICE) << this << ": query to localid not in overlay " << M->overlay_ << "@" << dst << " from " << src;
promise.set_error(td::Status::Error(ErrorCode::protoviolation, PSTRING() << "bad overlay_id " << M->overlay_));
return;
}
td::actor::send_closure(it2->second, &Overlay::update_throughput_in_ctr, src, (td::uint32)data.size(), true);
td::actor::send_closure(it2->second, &Overlay::receive_query, src, std::move(data), std::move(promise));
}
void OverlayManager::send_query_via(adnl::AdnlNodeIdShort dst, adnl::AdnlNodeIdShort src, OverlayIdShort overlay_id,
std::string name, td::Promise<td::BufferSlice> promise, td::Timestamp timeout,
td::BufferSlice query, td::uint64 max_answer_size,
td::actor::ActorId<adnl::AdnlSenderInterface> via) {
CHECK(query.size() <= adnl::Adnl::huge_packet_max_size());
auto it = overlays_.find(src);
if (it != overlays_.end()) {
auto it2 = it->second.find(overlay_id);
if (it2 != it->second.end()) {
td::actor::send_closure(it2->second, &Overlay::update_throughput_out_ctr, dst, (td::uint32)query.size(), true);
}
}
td::actor::send_closure(
via, &adnl::AdnlSenderInterface::send_query_ex, src, dst, std::move(name), std::move(promise), timeout,
create_serialize_tl_object_suffix<ton_api::overlay_query>(query.as_slice(), overlay_id.tl()), max_answer_size);
}
void OverlayManager::send_message_via(adnl::AdnlNodeIdShort dst, adnl::AdnlNodeIdShort src, OverlayIdShort overlay_id,
td::BufferSlice object, td::actor::ActorId<adnl::AdnlSenderInterface> via) {
CHECK(object.size() <= adnl::Adnl::huge_packet_max_size());
auto it = overlays_.find(src);
if (it != overlays_.end()) {
auto it2 = it->second.find(overlay_id);
if (it2 != it->second.end()) {
td::actor::send_closure(it2->second, &Overlay::update_throughput_out_ctr, dst, (td::uint32)object.size(), false);
}
}
td::actor::send_closure(
via, &adnl::AdnlSenderInterface::send_message, src, dst,
create_serialize_tl_object_suffix<ton_api::overlay_message>(object.as_slice(), overlay_id.tl()));
}
void OverlayManager::send_broadcast(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id, td::BufferSlice object) {
send_broadcast_ex(local_id, overlay_id, local_id.pubkey_hash(), 0, std::move(object));
}
void OverlayManager::send_broadcast_ex(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id, PublicKeyHash send_as,
td::uint32 flags, td::BufferSlice object) {
CHECK(object.size() <= Overlays::max_simple_broadcast_size());
auto it = overlays_.find(local_id);
if (it != overlays_.end()) {
auto it2 = it->second.find(overlay_id);
if (it2 != it->second.end()) {
td::actor::send_closure(it2->second, &Overlay::send_broadcast, send_as, flags, std::move(object));
}
}
}
void OverlayManager::send_broadcast_fec(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id,
td::BufferSlice object) {
send_broadcast_fec_ex(local_id, overlay_id, local_id.pubkey_hash(), 0, std::move(object));
}
void OverlayManager::send_broadcast_fec_ex(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id,
PublicKeyHash send_as, td::uint32 flags, td::BufferSlice object) {
CHECK(object.size() <= Overlays::max_fec_broadcast_size());
auto it = overlays_.find(local_id);
if (it != overlays_.end()) {
auto it2 = it->second.find(overlay_id);
if (it2 != it->second.end()) {
td::actor::send_closure(it2->second, &Overlay::send_broadcast_fec, send_as, flags, std::move(object));
}
}
}
void OverlayManager::set_privacy_rules(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id,
OverlayPrivacyRules rules) {
auto it = overlays_.find(local_id);
if (it != overlays_.end()) {
auto it2 = it->second.find(overlay_id);
if (it2 != it->second.end()) {
td::actor::send_closure(it2->second, &Overlay::set_privacy_rules, std::move(rules));
}
}
}
void OverlayManager::update_certificate(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id, PublicKeyHash key,
std::shared_ptr<Certificate> cert) {
auto it = overlays_.find(local_id);
if (it != overlays_.end()) {
auto it2 = it->second.find(overlay_id);
if (it2 != it->second.end()) {
td::actor::send_closure(it2->second, &Overlay::add_certificate, key, std::move(cert));
}
}
}
void OverlayManager::get_overlay_random_peers(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id,
td::uint32 max_peers,
td::Promise<std::vector<adnl::AdnlNodeIdShort>> promise) {
auto it = overlays_.find(local_id);
if (it != overlays_.end()) {
auto it2 = it->second.find(overlay_id);
if (it2 != it->second.end()) {
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,
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<dht::Dht> dht) {
return td::actor::create_actor<OverlayManager>("overlaymanager", db_root, keyring, adnl, dht);
}
OverlayManager::OverlayManager(std::string db_root, td::actor::ActorId<keyring::Keyring> keyring,
td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<dht::Dht> dht)
: db_root_(db_root), keyring_(keyring), adnl_(adnl), dht_node_(dht) {
}
void OverlayManager::start_up() {
std::shared_ptr<td::KeyValue> kv =
std::make_shared<td::RocksDb>(td::RocksDb::open(PSTRING() << db_root_ << "/overlays").move_as_ok());
db_ = DbType{std::move(kv)};
}
void OverlayManager::save_to_db(adnl::AdnlNodeIdShort local_id, OverlayIdShort overlay_id,
std::vector<OverlayNode> nodes) {
std::vector<tl_object_ptr<ton_api::overlay_node>> nodes_vec;
for (auto &n : nodes) {
nodes_vec.push_back(n.tl());
}
auto obj = create_tl_object<ton_api::overlay_nodes>(std::move(nodes_vec));
auto key = create_hash_tl_object<ton_api::overlay_db_key_nodes>(local_id.bits256_value(), overlay_id.bits256_value());
db_.set(key, create_serialize_tl_object<ton_api::overlay_db_nodes>(std::move(obj)));
}
void OverlayManager::get_stats(td::Promise<tl_object_ptr<ton_api::engine_validator_overlaysStats>> promise) {
class Cb : public td::actor::Actor {
public:
Cb(td::Promise<tl_object_ptr<ton_api::engine_validator_overlaysStats>> promise) : promise_(std::move(promise)) {
}
void incr_pending() {
pending_++;
}
void decr_pending() {
if (!--pending_) {
promise_.set_result(create_tl_object<ton_api::engine_validator_overlaysStats>(std::move(res_)));
stop();
}
}
void receive_answer(tl_object_ptr<ton_api::engine_validator_overlayStats> res) {
if (res) {
res_.push_back(std::move(res));
}
decr_pending();
}
private:
std::vector<tl_object_ptr<ton_api::engine_validator_overlayStats>> res_;
size_t pending_{1};
td::Promise<tl_object_ptr<ton_api::engine_validator_overlaysStats>> promise_;
};
auto act = td::actor::create_actor<Cb>("overlaysstatsmerger", std::move(promise)).release();
for (auto &a : overlays_) {
for (auto &b : a.second) {
td::actor::send_closure(act, &Cb::incr_pending);
td::actor::send_closure(b.second, &Overlay::get_stats,
[act](td::Result<tl_object_ptr<ton_api::engine_validator_overlayStats>> R) {
if (R.is_ok()) {
td::actor::send_closure(act, &Cb::receive_answer, R.move_as_ok());
} else {
td::actor::send_closure(act, &Cb::receive_answer, nullptr);
}
});
}
}
td::actor::send_closure(act, &Cb::decr_pending);
}
Certificate::Certificate(PublicKey issued_by, td::int32 expire_at, td::uint32 max_size, td::uint32 flags,
td::BufferSlice signature)
: issued_by_(issued_by)
, expire_at_(expire_at)
, max_size_(max_size)
, flags_(flags)
, signature_(td::SharedSlice(signature.as_slice())) {
}
Certificate::Certificate(PublicKeyHash issued_by, td::int32 expire_at, td::uint32 max_size, td::uint32 flags,
td::BufferSlice signature)
: issued_by_(issued_by)
, expire_at_(expire_at)
, max_size_(max_size)
, flags_(flags)
, signature_(td::SharedSlice(signature.as_slice())) {
}
void Certificate::set_signature(td::BufferSlice signature) {
signature_ = td::SharedSlice{signature.as_slice()};
}
void Certificate::set_issuer(PublicKey issuer) {
issued_by_ = issuer;
}
constexpr td::uint32 cert_default_flags(td::uint32 max_size) {
return (max_size > Overlays::max_simple_broadcast_size() ? CertificateFlags::AllowFec : 0) |
CertificateFlags::Trusted;
}
td::BufferSlice Certificate::to_sign(OverlayIdShort overlay_id, PublicKeyHash issued_to) const {
if (flags_ == cert_default_flags(max_size_)) {
return create_serialize_tl_object<ton_api::overlay_certificateId>(overlay_id.tl(), issued_to.tl(), expire_at_,
max_size_);
} else {
return create_serialize_tl_object<ton_api::overlay_certificateIdV2>(overlay_id.tl(), issued_to.tl(), expire_at_,
max_size_, flags_);
}
}
const PublicKeyHash Certificate::issuer_hash() const {
PublicKeyHash r;
issued_by_.visit(
td::overloaded([&](const PublicKeyHash &x) { r = x; }, [&](const PublicKey &x) { r = x.compute_short_id(); }));
return r;
}
const PublicKey &Certificate::issuer() const {
return issued_by_.get<PublicKey>();
}
td::Result<std::shared_ptr<Certificate>> Certificate::create(tl_object_ptr<ton_api::overlay_Certificate> cert) {
std::shared_ptr<Certificate> res;
ton_api::downcast_call(*cert.get(),
td::overloaded([&](ton_api::overlay_emptyCertificate &obj) { res = nullptr; },
[&](ton_api::overlay_certificate &obj) {
res = std::make_shared<Certificate>(PublicKey{obj.issued_by_}, obj.expire_at_,
static_cast<td::uint32>(obj.max_size_),
cert_default_flags(obj.max_size_),
std::move(obj.signature_));
},
[&](ton_api::overlay_certificateV2 &obj) {
res = std::make_shared<Certificate>(PublicKey{obj.issued_by_}, obj.expire_at_,
static_cast<td::uint32>(obj.max_size_),
static_cast<td::uint32>(obj.flags_),
std::move(obj.signature_));
}));
return std::move(res);
}
BroadcastCheckResult Certificate::check(PublicKeyHash node, OverlayIdShort overlay_id, td::int32 unix_time,
td::uint32 size, bool is_fec) const {
if (size > max_size_) {
return BroadcastCheckResult::Forbidden;
}
if (unix_time > expire_at_) {
return BroadcastCheckResult::Forbidden;
}
if (is_fec && !(flags_ & CertificateFlags::AllowFec)) {
return BroadcastCheckResult::Forbidden;
}
auto R1 = issued_by_.get<PublicKey>().create_encryptor();
if (R1.is_error()) {
return BroadcastCheckResult::Forbidden;
}
auto E = R1.move_as_ok();
auto B = to_sign(overlay_id, node);
if (E->check_signature(B.as_slice(), signature_.as_slice()).is_error()) {
return BroadcastCheckResult::Forbidden;
}
return (flags_ & CertificateFlags::Trusted) ? BroadcastCheckResult::Allowed : BroadcastCheckResult::NeedCheck;
}
tl_object_ptr<ton_api::overlay_Certificate> Certificate::tl() const {
return create_tl_object<ton_api::overlay_certificate>(issued_by_.get<PublicKey>().tl(), expire_at_, max_size_,
signature_.clone_as_buffer_slice());
}
tl_object_ptr<ton_api::overlay_Certificate> Certificate::empty_tl() {
return create_tl_object<ton_api::overlay_emptyCertificate>();
}
} // namespace overlay
} // namespace ton
Reference in a new issue View git blame Copy permalink