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
ton/storage/storage-daemon/StorageProvider.cpp
EmelyanenkoK 82e231d0a7
Fix deploy_storage_contract detection
2023-03-07 21:22:17 +03:00

862 lines
40 KiB
C++
Raw 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/>.
*/
#include "StorageProvider.h"
#include "td/db/RocksDb.h"
#include "td/utils/JsonBuilder.h"
#include "auto/tl/ton_api_json.h"
#include "td/utils/port/path.h"
#include "block/block-auto.h"
#include "common/delay.h"
#include "td/actor/MultiPromise.h"
namespace ton {
td::Result<ProviderParams> ProviderParams::create(const tl_object_ptr<ton_api::storage_daemon_provider_params>& obj) {
ProviderParams p;
p.accept_new_contracts = obj->accept_new_contracts_;
p.rate_per_mb_day = td::string_to_int256(obj->rate_per_mb_day_);
if (p.rate_per_mb_day.is_null() || p.rate_per_mb_day->sgn() < 0) {
return td::Status::Error("Invalid rate");
}
p.max_span = obj->max_span_;
p.minimal_file_size = obj->minimal_file_size_;
p.maximal_file_size = obj->maximal_file_size_;
return p;
}
tl_object_ptr<ton_api::storage_daemon_provider_params> ProviderParams::tl() const {
return create_tl_object<ton_api::storage_daemon_provider_params>(
accept_new_contracts, rate_per_mb_day->to_dec_string(), max_span, minimal_file_size, maximal_file_size);
}
bool ProviderParams::to_builder(vm::CellBuilder& b) const {
return b.store_long_bool(accept_new_contracts, 1) && store_coins(b, rate_per_mb_day) &&
b.store_long_bool(max_span, 32) && b.store_long_bool(minimal_file_size, 64) &&
b.store_long_bool(maximal_file_size, 64);
}
StorageProvider::StorageProvider(ContractAddress account_address, std::string db_root,
td::actor::ActorId<tonlib::TonlibClientWrapper> tonlib_client,
td::actor::ActorId<StorageManager> storage_manager,
td::actor::ActorId<keyring::Keyring> keyring)
: main_address_(account_address)
, db_root_(std::move(db_root))
, tonlib_client_(std::move(tonlib_client))
, storage_manager_(std::move(storage_manager))
, keyring_(std::move(keyring)) {
}
void StorageProvider::start_up() {
LOG(INFO) << "Initing storage provider, account address: " << main_address_.to_string();
td::mkdir(db_root_).ensure();
db_ = std::make_unique<td::RocksDb>(td::RocksDb::open(db_root_).move_as_ok());
auto r_state = db::db_get<ton_api::storage_provider_db_state>(
*db_, create_hash_tl_object<ton_api::storage_provider_db_key_state>(), true);
r_state.ensure();
auto state = r_state.move_as_ok();
if (state) {
last_processed_lt_ = state->last_processed_lt_;
LOG(INFO) << "Loaded storage provider state";
LOG(INFO) << "Last processed lt: " << last_processed_lt_;
}
class Callback : public FabricContractWrapper::Callback {
public:
explicit Callback(td::actor::ActorId<StorageProvider> id) : id_(std::move(id)) {
}
void on_transaction(tl_object_ptr<tonlib_api::raw_transaction> transaction) override {
td::actor::send_closure(id_, &StorageProvider::process_transaction, std::move(transaction));
}
private:
td::actor::ActorId<StorageProvider> id_;
};
contract_wrapper_ =
td::actor::create_actor<FabricContractWrapper>("ContractWrapper", main_address_, tonlib_client_, keyring_,
td::make_unique<Callback>(actor_id(this)), last_processed_lt_);
auto r_config = db::db_get<ton_api::storage_daemon_providerConfig>(
*db_, create_hash_tl_object<ton_api::storage_provider_db_key_providerConfig>(), true);
r_config.ensure();
auto config_obj = r_config.move_as_ok();
if (config_obj) {
LOG(INFO) << "Loaded config from db";
config_ = Config(config_obj);
} else {
LOG(INFO) << "Using default config";
db_store_config();
}
LOG(INFO) << "Config: max_contracts=" << config_.max_contracts << ", max_total_size=" << config_.max_total_size;
auto r_contract_list = db::db_get<ton_api::storage_provider_db_contractList>(
*db_, create_hash_tl_object<ton_api::storage_provider_db_key_contractList>(), true);
r_contract_list.ensure();
auto contract_list = r_contract_list.move_as_ok();
if (contract_list) {
LOG(INFO) << "Loading " << contract_list->contracts_.size() << " contracts from db";
for (auto& c : contract_list->contracts_) {
ContractAddress address(c->wc_, c->addr_);
if (contracts_.count(address)) {
LOG(ERROR) << "Duplicate contract in db: " << address.to_string();
continue;
}
auto r_contract = db::db_get<ton_api::storage_provider_db_storageContract>(
*db_, create_hash_tl_object<ton_api::storage_provider_db_key_storageContract>(address.wc, address.addr),
true);
r_contract.ensure();
auto db_contract = r_contract.move_as_ok();
if (!db_contract) {
LOG(ERROR) << "Missing contract in db: " << address.to_string();
continue;
}
StorageContract& contract = contracts_[address];
contract.torrent_hash = db_contract->torrent_hash_;
contract.microchunk_hash = db_contract->microchunk_hash_;
contract.created_time = db_contract->created_time_;
contract.state = (StorageContract::State)db_contract->state_;
contract.file_size = db_contract->file_size_;
contract.max_span = db_contract->max_span_;
contract.rate = td::string_to_int256(db_contract->rate_);
contracts_total_size_ += contract.file_size;
auto r_tree = db::db_get<ton_api::storage_provider_db_microchunkTree>(
*db_, create_hash_tl_object<ton_api::storage_provider_db_key_microchunkTree>(address.wc, address.addr), true);
r_tree.ensure();
auto tree = r_tree.move_as_ok();
if (tree) {
contract.microchunk_tree = std::make_shared<MicrochunkTree>(vm::std_boc_deserialize(tree->data_).move_as_ok());
}
LOG(INFO) << "Loaded contract from db: " << address.to_string() << ", torrent=" << contract.torrent_hash.to_hex()
<< ", state=" << contract.state;
}
}
for (auto& p : contracts_) {
const ContractAddress& address = p.first;
StorageContract& contract = p.second;
switch (contract.state) {
case StorageContract::st_downloading:
init_new_storage_contract(address, contract);
break;
case StorageContract::st_downloaded:
after_contract_downloaded(address);
break;
case StorageContract::st_active:
contract.check_next_proof_at = td::Timestamp::now();
break;
case StorageContract::st_closing:
check_storage_contract_deleted(address);
break;
default:
LOG(FATAL) << "Invalid contract state in db";
}
}
LOG(INFO) << "Loaded contracts from db";
alarm();
}
void StorageProvider::get_params(td::Promise<ProviderParams> promise) {
return get_provider_params(tonlib_client_, main_address_, std::move(promise));
}
void StorageProvider::get_provider_params(td::actor::ActorId<tonlib::TonlibClientWrapper> client,
ContractAddress address, td::Promise<ProviderParams> promise) {
run_get_method(
address, client, "get_storage_params", std::vector<tl_object_ptr<tonlib_api::tvm_StackEntry>>(),
promise.wrap([](std::vector<tl_object_ptr<tonlib_api::tvm_StackEntry>> stack) -> td::Result<ProviderParams> {
if (stack.size() != 5) {
return td::Status::Error(PSTRING() << "Method returned " << stack.size() << " values, 5 expected");
}
TRY_RESULT_PREFIX(accept_new_contracts, entry_to_int<int>(stack[0]), "Invalid accept_new_contracts: ");
TRY_RESULT_PREFIX(rate_per_mb_day, entry_to_int<td::RefInt256>(stack[1]), "Invalid rate_per_mb_day: ");
TRY_RESULT_PREFIX(max_span, entry_to_int<td::uint32>(stack[2]), "Invalid max_span: ");
TRY_RESULT_PREFIX(minimal_file_size, entry_to_int<td::uint64>(stack[3]), "Invalid minimal_file_size: ");
TRY_RESULT_PREFIX(maximal_file_size, entry_to_int<td::uint64>(stack[4]), "Invalid maximal_file_size: ");
ProviderParams params;
params.accept_new_contracts = accept_new_contracts;
params.rate_per_mb_day = rate_per_mb_day;
params.max_span = max_span;
params.minimal_file_size = minimal_file_size;
params.maximal_file_size = maximal_file_size;
return params;
}));
}
void StorageProvider::set_params(ProviderParams params, td::Promise<td::Unit> promise) {
vm::CellBuilder b;
b.store_long(0x54cbf19b, 32); // const op::update_storage_params = 0x54cbf19b;
b.store_long(0, 64); // query_id
if (!params.to_builder(b)) {
promise.set_error(td::Status::Error("Failed to store params to builder"));
return;
}
LOG(INFO) << "Sending external message to update provider parameters: " << params.accept_new_contracts << ", "
<< params.max_span << ", " << params.rate_per_mb_day << ", " << params.minimal_file_size << ", "
<< params.maximal_file_size;
td::actor::send_closure(contract_wrapper_, &FabricContractWrapper::send_internal_message, main_address_,
td::make_refint(100'000'000), b.as_cellslice(), std::move(promise));
}
void StorageProvider::db_store_state() {
LOG(DEBUG) << "db_store_state last_lt=" << last_processed_lt_;
db_->begin_transaction().ensure();
db_->set(create_hash_tl_object<ton_api::storage_provider_db_key_state>().as_slice(),
create_serialize_tl_object<ton_api::storage_provider_db_state>(last_processed_lt_))
.ensure();
db_->commit_transaction().ensure();
}
void StorageProvider::db_store_config() {
LOG(DEBUG) << "db_store_config";
db_->begin_transaction().ensure();
db_->set(create_hash_tl_object<ton_api::storage_provider_db_key_providerConfig>().as_slice(),
serialize_tl_object(config_.tl(), true))
.ensure();
db_->commit_transaction().ensure();
}
void StorageProvider::alarm() {
for (auto& p : contracts_) {
if (p.second.check_next_proof_at && p.second.check_next_proof_at.is_in_past()) {
p.second.check_next_proof_at = td::Timestamp::never();
check_next_proof(p.first, p.second);
}
alarm_timestamp().relax(p.second.check_next_proof_at);
}
}
void StorageProvider::process_transaction(tl_object_ptr<tonlib_api::raw_transaction> transaction) {
std::string new_contract_address;
for (auto& message : transaction->out_msgs_) {
auto data = dynamic_cast<tonlib_api::msg_dataRaw*>(message->msg_data_.get());
if (data == nullptr) {
continue;
}
auto r_body = vm::std_boc_deserialize(data->body_);
if (r_body.is_error()) {
LOG(ERROR) << "Invalid message body in tonlib response: " << r_body.move_as_error();
continue;
}
td::Ref<vm::Cell> body = r_body.move_as_ok();
vm::CellSlice cs = vm::load_cell_slice(body);
if (cs.size() >= 32) {
long long op_code = cs.prefetch_ulong(32);
// const op::offer_storage_contract = 0x107c49ef; -- old versions
// const op::deploy_storage_contract = 0xe4748df1; -- new versions
if((op_code == 0x107c49ef) || (op_code == 0xe4748df1)) {
new_contract_address = message->destination_->account_address_;
}
}
}
if (!new_contract_address.empty()) {
auto P = td::PromiseCreator::lambda(
[SelfId = actor_id(this), lt = (td::uint64)transaction->transaction_id_->lt_](td::Result<td::Unit> R) {
if (R.is_error()) {
LOG(ERROR) << "Error during processing new storage contract, skipping: " << R.move_as_error();
}
});
on_new_storage_contract(ContractAddress::parse(new_contract_address).move_as_ok(), std::move(P));
}
last_processed_lt_ = transaction->transaction_id_->lt_;
db_store_state();
}
void StorageProvider::on_new_storage_contract(ContractAddress address, td::Promise<td::Unit> promise, int max_retries) {
LOG(INFO) << "Processing new storage contract: " << address.to_string();
get_storage_contract_data(
address, tonlib_client_,
[SelfId = actor_id(this), address, promise = std::move(promise),
max_retries](td::Result<StorageContractData> R) mutable {
if (R.is_error()) {
if (max_retries > 0) {
LOG(WARNING) << "Processing new storage contract: " << R.move_as_error() << ", retrying";
delay_action(
[SelfId = std::move(SelfId), promise = std::move(promise), address = std::move(address),
max_retries]() mutable {
td::actor::send_closure(SelfId, &StorageProvider::on_new_storage_contract, std::move(address),
std::move(promise), max_retries - 1);
},
td::Timestamp::in(5.0));
} else {
promise.set_error(R.move_as_error());
}
return;
}
td::actor::send_closure(SelfId, &StorageProvider::on_new_storage_contract_cont, address, R.move_as_ok(),
std::move(promise));
});
}
void StorageProvider::on_new_storage_contract_cont(ContractAddress address, StorageContractData data,
td::Promise<td::Unit> promise) {
auto it = contracts_.emplace(address, StorageContract());
if (!it.second) {
promise.set_error(td::Status::Error(PSTRING() << "Storage contract already registered: " << address.to_string()));
return;
}
LOG(INFO) << "New storage contract " << address.to_string() << ", torrent hash: " << data.torrent_hash.to_hex();
LOG(DEBUG) << "Stoage contract data: microchunk_hash=" << data.microchunk_hash << ", balance=" << data.balance
<< ", file_size=" << data.file_size << ", next_proof=" << data.next_proof
<< ", rate=" << data.rate_per_mb_day << ", max_span=" << data.max_span;
StorageContract& contract = it.first->second;
contract.torrent_hash = data.torrent_hash;
contract.microchunk_hash = data.microchunk_hash;
contract.state = StorageContract::st_downloading;
contract.created_time = (td::uint32)td::Clocks::system();
contract.file_size = data.file_size;
contract.max_span = data.max_span;
contract.rate = data.rate_per_mb_day;
contracts_total_size_ += contract.file_size;
promise.set_result(td::Unit());
if (contracts_.size() <= config_.max_contracts && contracts_total_size_ <= config_.max_total_size) {
db_update_storage_contract(address, true);
init_new_storage_contract(address, contract);
} else {
if (contracts_.size() > config_.max_contracts) {
LOG(WARNING) << "Cannot add new storage contract: too many contracts (limit = " << config_.max_contracts << ")";
} else {
LOG(WARNING) << "Cannot add new storage contract: total size exceeded (limit = "
<< td::format::as_size(config_.max_total_size) << ")";
}
contract.state = StorageContract::st_closing;
db_update_storage_contract(address, true);
do_close_storage_contract(address);
}
}
void StorageProvider::db_update_storage_contract(const ContractAddress& address, bool update_list) {
LOG(DEBUG) << "db_update_storage_contract " << address.to_string() << " " << update_list;
db_->begin_transaction().ensure();
if (update_list) {
std::vector<tl_object_ptr<ton_api::storage_provider_db_contractAddress>> list;
for (const auto& t : contracts_) {
list.push_back(create_tl_object<ton_api::storage_provider_db_contractAddress>(t.first.wc, t.first.addr));
}
db_->set(create_hash_tl_object<ton_api::storage_provider_db_key_contractList>().as_slice(),
create_serialize_tl_object<ton_api::storage_provider_db_contractList>(std::move(list)))
.ensure();
}
auto key = create_hash_tl_object<ton_api::storage_provider_db_key_storageContract>(address.wc, address.addr);
auto it = contracts_.find(address);
if (it == contracts_.end()) {
db_->erase(key.as_slice()).ensure();
} else {
const StorageContract& contract = it->second;
db_->set(key.as_slice(),
create_serialize_tl_object<ton_api::storage_provider_db_storageContract>(
contract.torrent_hash, contract.microchunk_hash, contract.created_time, (int)contract.state,
contract.file_size, contract.rate->to_dec_string(), contract.max_span));
}
db_->commit_transaction().ensure();
}
void StorageProvider::db_update_microchunk_tree(const ContractAddress& address) {
LOG(DEBUG) << "db_update_microchunk_tree " << address.to_string();
db_->begin_transaction().ensure();
auto key = create_hash_tl_object<ton_api::storage_provider_db_key_microchunkTree>(address.wc, address.addr);
auto it = contracts_.find(address);
if (it == contracts_.end() || it->second.microchunk_tree == nullptr) {
db_->erase(key.as_slice()).ensure();
} else {
db_->set(key.as_slice(), create_serialize_tl_object<ton_api::storage_provider_db_microchunkTree>(
vm::std_boc_serialize(it->second.microchunk_tree->get_root()).move_as_ok()));
}
db_->commit_transaction().ensure();
}
void StorageProvider::init_new_storage_contract(ContractAddress address, StorageContract& contract) {
CHECK(contract.state == StorageContract::st_downloading);
td::actor::send_closure(storage_manager_, &StorageManager::add_torrent_by_hash, contract.torrent_hash, "", false,
false, [](td::Result<td::Unit> R) {
// Ignore errors: error can mean that the torrent already exists, other errors will be caught later
if (R.is_error()) {
LOG(DEBUG) << "Add torrent: " << R.move_as_error();
} else {
LOG(DEBUG) << "Add torrent: OK";
}
});
td::actor::send_closure(storage_manager_, &StorageManager::set_active_download, contract.torrent_hash, true,
[SelfId = actor_id(this), address](td::Result<td::Unit> R) {
if (R.is_error()) {
LOG(ERROR) << "Failed to init storage contract: " << R.move_as_error();
td::actor::send_closure(SelfId, &StorageProvider::do_close_storage_contract, address);
return;
}
LOG(DEBUG) << "Set active download: OK";
});
td::actor::send_closure(
storage_manager_, &StorageManager::wait_for_completion, contract.torrent_hash,
[SelfId = actor_id(this), address, hash = contract.torrent_hash, microchunk_hash = contract.microchunk_hash,
manager = storage_manager_](td::Result<td::Unit> R) {
if (R.is_error()) {
LOG(WARNING) << "Failed to download torrent " << hash.to_hex() << ": " << R.move_as_error();
td::actor::send_closure(SelfId, &StorageProvider::do_close_storage_contract, address);
return;
}
LOG(DEBUG) << "Downloaded torrent " << hash;
td::actor::send_closure(
manager, &StorageManager::with_torrent, hash,
[SelfId, address, hash, microchunk_hash](td::Result<NodeActor::NodeState> R) {
auto r_microchunk_tree = [&]() -> td::Result<MicrochunkTree> {
TRY_RESULT(state, std::move(R));
Torrent& torrent = state.torrent;
if (!torrent.is_completed() || torrent.get_included_size() != torrent.get_info().file_size) {
return td::Status::Error("unknown error");
}
LOG(DEBUG) << "Building microchunk tree for " << hash;
TRY_RESULT(tree, MicrochunkTree::Builder::build_for_torrent(torrent));
if (tree.get_root_hash() != microchunk_hash) {
return td::Status::Error("microchunk tree hash mismatch");
}
return tree;
}();
if (r_microchunk_tree.is_error()) {
LOG(WARNING) << "Failed to download torrent " << hash.to_hex() << ": " << R.move_as_error();
td::actor::send_closure(SelfId, &StorageProvider::do_close_storage_contract, address);
} else {
td::actor::send_closure(SelfId, &StorageProvider::downloaded_torrent, address,
r_microchunk_tree.move_as_ok());
}
});
});
}
void StorageProvider::downloaded_torrent(ContractAddress address, MicrochunkTree microchunk_tree) {
auto it = contracts_.find(address);
if (it == contracts_.end()) {
LOG(WARNING) << "Contract " << address.to_string() << " does not exist anymore";
return;
}
auto& contract = it->second;
LOG(INFO) << "Finished downloading torrent " << contract.torrent_hash.to_hex() << " for contract "
<< address.to_string();
contract.state = StorageContract::st_downloaded;
contract.microchunk_tree = std::make_shared<MicrochunkTree>(std::move(microchunk_tree));
db_update_microchunk_tree(address);
db_update_storage_contract(address, false);
after_contract_downloaded(address);
}
void StorageProvider::after_contract_downloaded(ContractAddress address, td::Timestamp retry_until,
td::Timestamp retry_false_until) {
auto it = contracts_.find(address);
if (it == contracts_.end()) {
LOG(WARNING) << "Contract " << address.to_string() << " does not exist anymore";
return;
}
auto& contract = it->second;
td::actor::send_closure(storage_manager_, &StorageManager::set_active_upload, contract.torrent_hash, true,
[SelfId = actor_id(this), address](td::Result<td::Unit> R) {
if (R.is_error()) {
LOG(ERROR) << "Set active upload: " << R.move_as_error();
return;
}
LOG(DEBUG) << "Set active upload: OK";
});
get_storage_contract_data(address, tonlib_client_,
[=, SelfId = actor_id(this)](td::Result<StorageContractData> R) mutable {
if (R.is_error()) {
LOG(WARNING) << "Failed to check that contract is active: " << R.move_as_error();
if (retry_until && retry_until.is_in_past()) {
delay_action(
[=]() {
td::actor::send_closure(SelfId, &StorageProvider::after_contract_downloaded,
address, retry_until, retry_false_until);
},
td::Timestamp::in(5.0));
}
return;
}
if (R.ok().active) {
td::actor::send_closure(SelfId, &StorageProvider::activated_storage_contract, address);
} else if (retry_false_until && retry_false_until.is_in_past()) {
delay_action(
[=]() {
td::actor::send_closure(SelfId, &StorageProvider::after_contract_downloaded,
address, retry_until, retry_false_until);
},
td::Timestamp::in(5.0));
} else {
td::actor::send_closure(SelfId, &StorageProvider::activate_contract_cont, address);
}
});
}
void StorageProvider::activate_contract_cont(ContractAddress address) {
vm::CellBuilder b;
b.store_long(0x7a361688, 32); // const op::accept_storage_contract = 0x7a361688;
b.store_long(0, 64); // query_id
LOG(DEBUG) << "Sending op::accept_storage_contract to " << address.to_string();
td::actor::send_closure(
contract_wrapper_, &FabricContractWrapper::send_internal_message, address, td::make_refint(100'000'000),
b.as_cellslice(), [SelfId = actor_id(this), address](td::Result<td::Unit> R) {
if (R.is_error()) {
LOG(ERROR) << "Failed to send activate message, retrying later: " << R.move_as_error();
delay_action([=]() { td::actor::send_closure(SelfId, &StorageProvider::activate_contract_cont, address); },
td::Timestamp::in(10.0));
return;
}
td::actor::send_closure(SelfId, &StorageProvider::after_contract_downloaded, address, td::Timestamp::in(60.0),
td::Timestamp::in(40.0));
});
}
void StorageProvider::activated_storage_contract(ContractAddress address) {
auto it = contracts_.find(address);
if (it == contracts_.end()) {
LOG(WARNING) << "Contract " << address.to_string() << " does not exist anymore";
return;
}
LOG(INFO) << "Storage contract " << address.to_string() << " is active";
auto& contract = it->second;
contract.state = StorageContract::st_active;
db_update_storage_contract(address, false);
alarm_timestamp().relax(contract.check_next_proof_at = td::Timestamp::in(1.0));
}
void StorageProvider::do_close_storage_contract(ContractAddress address) {
auto it = contracts_.find(address);
if (it == contracts_.end()) {
LOG(WARNING) << "Contract " << address.to_string() << " does not exist anymore";
return;
}
LOG(INFO) << "Closing storage contract " << address.to_string();
auto& contract = it->second;
contract.state = StorageContract::st_closing;
db_update_storage_contract(address, false);
check_storage_contract_deleted(address);
}
void StorageProvider::send_close_storage_contract(ContractAddress address) {
vm::CellBuilder b;
b.store_long(0x79f937ea, 32); // const op::close_contract = 0x79f937ea;
b.store_long(0, 64); // query_id
LOG(DEBUG) << "Sending op::close_contract to " << address.to_string();
td::actor::send_closure(
contract_wrapper_, &FabricContractWrapper::send_internal_message, address, td::make_refint(100'000'000),
b.as_cellslice(), [SelfId = actor_id(this), address](td::Result<td::Unit> R) {
if (R.is_error()) {
LOG(ERROR) << "Failed to send close message, retrying later: " << R.move_as_error();
delay_action([=]() { td::actor::send_closure(SelfId, &StorageProvider::activate_contract_cont, address); },
td::Timestamp::in(10.0));
return;
}
td::actor::send_closure(SelfId, &StorageProvider::check_storage_contract_deleted, address,
td::Timestamp::in(40.0));
});
}
void StorageProvider::check_storage_contract_deleted(ContractAddress address, td::Timestamp retry_false_until) {
check_contract_exists(address, tonlib_client_, [=, SelfId = actor_id(this)](td::Result<bool> R) {
if (R.is_error()) {
delay_action(
[=]() {
td::actor::send_closure(SelfId, &StorageProvider::check_storage_contract_deleted, address,
retry_false_until);
},
td::Timestamp::in(10.0));
return;
}
if (!R.move_as_ok()) {
td::actor::send_closure(SelfId, &StorageProvider::storage_contract_deleted, address);
} else if (retry_false_until && !retry_false_until.is_in_past()) {
delay_action(
[=]() {
td::actor::send_closure(SelfId, &StorageProvider::check_storage_contract_deleted, address,
retry_false_until);
},
td::Timestamp::in(5.0));
} else {
td::actor::send_closure(SelfId, &StorageProvider::send_close_storage_contract, address);
}
});
}
void StorageProvider::storage_contract_deleted(ContractAddress address) {
auto it = contracts_.find(address);
if (it == contracts_.end()) {
return;
}
LOG(INFO) << "Storage contract " << address.to_string() << " was deleted";
td::Bits256 hash = it->second.torrent_hash;
contracts_total_size_ -= it->second.file_size;
contracts_.erase(it);
bool delete_torrent = true;
for (const auto& p : contracts_) {
if (p.second.torrent_hash == hash) {
delete_torrent = false;
break;
}
}
if (delete_torrent) {
LOG(INFO) << "Deleting torrent " << hash.to_hex();
td::actor::send_closure(storage_manager_, &StorageManager::remove_torrent, hash, true,
[](td::Result<td::Unit> R) {});
}
db_update_storage_contract(address, true);
}
void StorageProvider::check_next_proof(ContractAddress address, StorageContract& contract) {
if (contract.state != StorageContract::st_active) {
return;
}
CHECK(contract.microchunk_tree != nullptr);
get_storage_contract_data(
address, tonlib_client_, [SelfId = actor_id(this), address](td::Result<StorageContractData> R) {
td::actor::send_closure(SelfId, &StorageProvider::got_next_proof_info, address, std::move(R));
});
}
void StorageProvider::got_next_proof_info(ContractAddress address, td::Result<StorageContractData> R) {
auto it = contracts_.find(address);
if (it == contracts_.end() || it->second.state != StorageContract::st_active) {
return;
}
auto& contract = it->second;
if (R.is_error()) {
LOG(ERROR) << "get_next_proof_info for " << address.to_string() << ": " << R.move_as_error();
check_contract_exists(address, tonlib_client_, [SelfId = actor_id(this), address](td::Result<bool> R) {
td::actor::send_closure(SelfId, &StorageProvider::got_contract_exists, address, std::move(R));
});
return;
}
auto data = R.move_as_ok();
if (data.balance->sgn() == 0) {
LOG(INFO) << "Balance of contract " << address.to_string() << " is zero, closing";
do_close_storage_contract(address);
return;
}
td::uint32 send_at = data.last_proof_time + data.max_span / 2, now = (td::uint32)td::Clocks::system();
if (now < send_at) {
LOG(DEBUG) << "Will send proof in " << send_at - now << "s (last_proof_time=" << data.last_proof_time
<< ", max_span=" << data.max_span << ")";
alarm_timestamp().relax(contract.check_next_proof_at = td::Timestamp::in(send_at - now + 2));
return;
}
LOG(INFO) << "Sending proof for " << address.to_string() << ": next_proof=" << data.next_proof
<< ", max_span=" << data.max_span << ", last_proof_time=" << data.last_proof_time << " ("
<< now - data.last_proof_time << "s ago)";
td::actor::send_closure(
storage_manager_, &StorageManager::with_torrent, contract.torrent_hash,
[=, SelfId = actor_id(this), tree = contract.microchunk_tree](td::Result<NodeActor::NodeState> R) {
if (R.is_error()) {
LOG(ERROR) << "Missing torrent for " << address.to_string();
return;
}
auto state = R.move_as_ok();
td::uint64 l = data.next_proof / MicrochunkTree::MICROCHUNK_SIZE * MicrochunkTree::MICROCHUNK_SIZE;
td::uint64 r = l + MicrochunkTree::MICROCHUNK_SIZE;
auto proof = tree->get_proof(l, r, state.torrent);
td::actor::send_closure(SelfId, &StorageProvider::got_next_proof, address, std::move(proof));
});
}
void StorageProvider::got_contract_exists(ContractAddress address, td::Result<bool> R) {
auto it = contracts_.find(address);
if (it == contracts_.end() || it->second.state != StorageContract::st_active) {
return;
}
auto& contract = it->second;
if (R.is_error()) {
LOG(ERROR) << "Check contract exists for " << address.to_string() << ": " << R.move_as_error();
alarm_timestamp().relax(contract.check_next_proof_at = td::Timestamp::in(10.0));
return;
}
if (R.ok()) {
alarm_timestamp().relax(contract.check_next_proof_at = td::Timestamp::in(10.0));
return;
}
storage_contract_deleted(address);
}
void StorageProvider::got_next_proof(ContractAddress address, td::Result<td::Ref<vm::Cell>> R) {
if (R.is_error()) {
LOG(ERROR) << "Failed to build proof: " << R.move_as_error();
return;
}
LOG(INFO) << "Got proof, sending";
vm::CellBuilder b;
b.store_long(0x419d5d4d, 32); // const op::proof_storage = 0x419d5d4d;
b.store_long(0, 64); // query_id
b.store_ref(R.move_as_ok());
td::actor::send_closure(contract_wrapper_, &FabricContractWrapper::send_internal_message, address,
td::make_refint(100'000'000), b.as_cellslice(),
[SelfId = actor_id(this), address](td::Result<td::Unit> R) {
if (R.is_error()) {
LOG(ERROR) << "Failed to send proof message: " << R.move_as_error();
} else {
LOG(DEBUG) << "Proof for " << address.to_string() << " was sent";
}
td::actor::send_closure(SelfId, &StorageProvider::sent_next_proof, address);
});
}
void StorageProvider::sent_next_proof(ContractAddress address) {
auto it = contracts_.find(address);
if (it == contracts_.end() || it->second.state != StorageContract::st_active) {
return;
}
auto& contract = it->second;
alarm_timestamp().relax(contract.check_next_proof_at = td::Timestamp::in(30.0));
}
void StorageProvider::get_provider_info(bool with_balances, bool with_contracts,
td::Promise<tl_object_ptr<ton_api::storage_daemon_providerInfo>> promise) {
auto result = std::make_shared<ton_api::storage_daemon_providerInfo>();
td::MultiPromise mp;
auto ig = mp.init_guard();
ig.add_promise(promise.wrap(
[result](td::Unit) { return create_tl_object<ton_api::storage_daemon_providerInfo>(std::move(*result)); }));
result->address_ = main_address_.to_string();
result->config_ = config_.tl();
result->contracts_count_ = (int)contracts_.size();
result->contracts_total_size_ = contracts_total_size_;
if (with_balances) {
get_contract_balance(main_address_, tonlib_client_, ig.get_promise().wrap([result](td::RefInt256 balance) {
result->balance_ = balance->to_dec_string();
return td::Unit();
}));
} else {
result->balance_ = "-1";
}
if (with_contracts) {
for (const auto& p : contracts_) {
auto obj = create_tl_object<ton_api::storage_daemon_contractInfo>();
const StorageContract& contract = p.second;
obj->address_ = p.first.to_string();
obj->state_ = (int)contract.state;
obj->torrent_ = contract.torrent_hash;
obj->created_time_ = contract.created_time;
obj->rate_ = contract.rate->to_dec_string();
obj->max_span_ = contract.max_span;
obj->file_size_ = contract.file_size;
obj->downloaded_size_ = obj->file_size_;
obj->client_balance_ = "-1";
obj->contract_balance_ = "-1";
result->contracts_.push_back(std::move(obj));
}
size_t i = 0;
for (const auto& p : contracts_) {
const StorageContract& contract = p.second;
if (contract.state == StorageContract::st_downloading) {
td::actor::send_closure(storage_manager_, &StorageManager::with_torrent, contract.torrent_hash,
[i, result, promise = ig.get_promise()](td::Result<NodeActor::NodeState> R) mutable {
if (R.is_error()) {
result->contracts_[i]->downloaded_size_ = 0;
} else {
auto state = R.move_as_ok();
result->contracts_[i]->downloaded_size_ = state.torrent.get_included_ready_size();
}
promise.set_result(td::Unit());
});
}
if (with_balances) {
get_contract_balance(p.first, tonlib_client_,
[i, result, promise = ig.get_promise()](td::Result<td::RefInt256> R) mutable {
if (R.is_ok()) {
result->contracts_[i]->contract_balance_ = R.ok()->to_dec_string();
}
promise.set_result(td::Unit());
});
get_storage_contract_data(p.first, tonlib_client_,
[i, result, promise = ig.get_promise()](td::Result<StorageContractData> R) mutable {
auto S = [&]() -> td::Status {
TRY_RESULT(data, std::move(R));
result->contracts_[i]->client_balance_ = data.balance->to_dec_string();
return td::Status::OK();
}();
promise.set_result(td::Unit());
});
}
i += 1;
}
}
}
void StorageProvider::set_provider_config(Config config, td::Promise<td::Unit> promise) {
config_ = config;
LOG(INFO) << "Changing provider config: max_contracts=" << config_.max_contracts
<< ", max_total_size=" << config_.max_total_size;
db_store_config();
promise.set_result(td::Unit());
}
void StorageProvider::withdraw(ContractAddress address, td::Promise<td::Unit> promise) {
auto it = contracts_.find(address);
if (it == contracts_.end() || it->second.state != StorageContract::st_active) {
promise.set_error(td::Status::Error("No such storage contract"));
return;
}
if (it->second.state != StorageContract::st_active) {
promise.set_error(td::Status::Error("Storage contract is not active"));
return;
}
vm::CellBuilder b;
b.store_long(0x46ed2e94, 32); // const op::withdraw = 0x46ed2e94;
b.store_long(0, 64); // query_id
LOG(INFO) << "Sending op::withdraw to storage contract " << address.to_string();
td::actor::send_closure(contract_wrapper_, &FabricContractWrapper::send_internal_message, address,
td::make_refint(100'000'000), b.as_cellslice(), std::move(promise));
}
void StorageProvider::send_coins(ContractAddress dest, td::RefInt256 amount, std::string message,
td::Promise<td::Unit> promise) {
if (amount->sgn() < 0) {
promise.set_error(td::Status::Error("Amount is negative"));
return;
}
vm::CellBuilder b;
if (!message.empty()) {
b.store_long(0, 32);
if (b.remaining_bits() < message.size() * 8) {
promise.set_error(td::Status::Error("Message is too long (max 122 bytes)"));
return;
}
b.store_bytes(td::Slice(message));
}
LOG(INFO) << "Sending " << amount << " nanoTON to " << dest.to_string();
td::actor::send_closure(contract_wrapper_, &FabricContractWrapper::send_internal_message, dest, amount,
b.finalize_novm(), std::move(promise));
}
void StorageProvider::close_storage_contract(ContractAddress address, td::Promise<td::Unit> promise) {
if (!contracts_.count(address)) {
promise.set_error(td::Status::Error("No such storage contract"));
return;
}
do_close_storage_contract(address);
promise.set_result(td::Unit());
}
StorageProvider::Config::Config(const tl_object_ptr<ton_api::storage_daemon_providerConfig>& obj)
: max_contracts(obj->max_contracts_), max_total_size(obj->max_total_size_) {
}
tl_object_ptr<ton_api::storage_daemon_providerConfig> StorageProvider::Config::tl() const {
return create_tl_object<ton_api::storage_daemon_providerConfig>(max_contracts, max_total_size);
}
} // namespace ton
Reference in a new issue View git blame Copy permalink