@ -68,10 +68,12 @@
#define BLOCK_QUEUE_FORCE_DOWNLOAD_NEAR_BLOCKS 1000
#define BLOCK_QUEUE_FORCE_DOWNLOAD_NEAR_BLOCKS 1000
#define REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD_STANDBY (5 * 1000000) // microseconds
#define REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD_STANDBY (5 * 1000000) // microseconds
#define REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD (30 * 1000000) // microseconds
#define REQUEST_NEXT_SCHEDULED_SPAN_THRESHOLD (30 * 1000000) // microseconds
#define IDLE_PEER_KICK_TIME (600 * 1000000) // microseconds
#define IDLE_PEER_KICK_TIME (240 * 1000000) // microseconds
#define NON_RESPONSIVE_PEER_KICK_TIME (20 * 1000000) // microseconds
#define PASSIVE_PEER_KICK_TIME (60 * 1000000) // microseconds
#define PASSIVE_PEER_KICK_TIME (60 * 1000000) // microseconds
#define DROP_ON_SYNC_WEDGE_THRESHOLD (30 * 1000000000ull) // nanoseconds
#define DROP_ON_SYNC_WEDGE_THRESHOLD (30 * 1000000000ull) // nanoseconds
#define LAST_ACTIVITY_STALL_THRESHOLD (2.0f) // seconds
#define LAST_ACTIVITY_STALL_THRESHOLD (2.0f) // seconds
#define DROP_PEERS_ON_SCORE -2
namespace cryptonote
namespace cryptonote
{
{
@ -139,9 +141,12 @@ namespace cryptonote
{
{
NOTIFY_REQUEST_CHAIN::request r = {};
NOTIFY_REQUEST_CHAIN::request r = {};
context.m_needed_objects.clear();
context.m_needed_objects.clear();
context.m_expect_height = m_core.get_current_blockchain_height();
m_core.get_short_chain_history(r.block_ids);
m_core.get_short_chain_history(r.block_ids);
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
r.prune = m_sync_pruned_blocks;
r.prune = m_sync_pruned_blocks;
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
context.m_expect_response = NOTIFY_RESPONSE_CHAIN_ENTRY::ID;
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() );
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() );
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
MLOG_PEER_STATE("requesting chain");
MLOG_PEER_STATE("requesting chain");
@ -326,6 +331,11 @@ namespace cryptonote
}
}
}
}
if (hshd.current_height < context.m_remote_blockchain_height)
{
MINFO(context << "Claims " << hshd.current_height << ", claimed " << context.m_remote_blockchain_height << " before");
hit_score(context, 1);
}
context.m_remote_blockchain_height = hshd.current_height;
context.m_remote_blockchain_height = hshd.current_height;
context.m_pruning_seed = hshd.pruning_seed;
context.m_pruning_seed = hshd.pruning_seed;
#ifdef CRYPTONOTE_PRUNING_DEBUG_SPOOF_SEED
#ifdef CRYPTONOTE_PRUNING_DEBUG_SPOOF_SEED
@ -416,7 +426,7 @@ namespace cryptonote
}
}
//------------------------------------------------------------------------------------------------------------------------
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
template<class t_core>
bool t_cryptonote_protocol_handler<t_core>::get_payload_sync_data(blobdata & data)
bool t_cryptonote_protocol_handler<t_core>::get_payload_sync_data(epee::byte_slice & data)
{
{
CORE_SYNC_DATA hsd = {};
CORE_SYNC_DATA hsd = {};
get_payload_sync_data(hsd);
get_payload_sync_data(hsd);
@ -427,10 +437,10 @@ namespace cryptonote
template<class t_core>
template<class t_core>
int t_cryptonote_protocol_handler<t_core>::handle_notify_new_block(int command, NOTIFY_NEW_BLOCK::request& arg, cryptonote_connection_context& context)
int t_cryptonote_protocol_handler<t_core>::handle_notify_new_block(int command, NOTIFY_NEW_BLOCK::request& arg, cryptonote_connection_context& context)
{
{
MLOGIF_P2P_MESSAGE(crypto::hash hash; cryptonote::block b; bool ret = cryptonote::parse_and_validate_block_from_blob(arg.b.block, b, &hash);, ret, "Received NOTIFY_NEW_BLOCK " << hash << " (height " << arg.current_blockchain_height << ", " << arg.b.txs.size() << " txes)");
MLOGIF_P2P_MESSAGE(crypto::hash hash; cryptonote::block b; bool ret = cryptonote::parse_and_validate_block_from_blob(arg.b.block, b, &hash);, ret, context << "Received NOTIFY_NEW_BLOCK " << hash << " (height " << arg.current_blockchain_height << ", " << arg.b.txs.size() << " txes)");
if(context.m_state != cryptonote_connection_context::state_normal)
if(context.m_state != cryptonote_connection_context::state_normal)
return 1;
return 1;
if(!is_synchronized() || m_no_sync ) // can happen if a peer connection goes to normal but another thread still hasn't finished adding queued blocks
if(!is_synchronized()) // can happen if a peer connection goes to normal but another thread still hasn't finished adding queued blocks
{
{
LOG_DEBUG_CC(context, "Received new block while syncing, ignored");
LOG_DEBUG_CC(context, "Received new block while syncing, ignored");
return 1;
return 1;
@ -484,9 +494,12 @@ namespace cryptonote
context.m_needed_objects.clear();
context.m_needed_objects.clear();
context.m_state = cryptonote_connection_context::state_synchronizing;
context.m_state = cryptonote_connection_context::state_synchronizing;
NOTIFY_REQUEST_CHAIN::request r = {};
NOTIFY_REQUEST_CHAIN::request r = {};
context.m_expect_height = m_core.get_current_blockchain_height();
m_core.get_short_chain_history(r.block_ids);
m_core.get_short_chain_history(r.block_ids);
r.prune = m_sync_pruned_blocks;
r.prune = m_sync_pruned_blocks;
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
context.m_expect_response = NOTIFY_RESPONSE_CHAIN_ENTRY::ID;
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() );
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() );
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
MLOG_PEER_STATE("requesting chain");
MLOG_PEER_STATE("requesting chain");
@ -498,10 +511,10 @@ namespace cryptonote
template<class t_core>
template<class t_core>
int t_cryptonote_protocol_handler<t_core>::handle_notify_new_fluffy_block(int command, NOTIFY_NEW_FLUFFY_BLOCK::request& arg, cryptonote_connection_context& context)
int t_cryptonote_protocol_handler<t_core>::handle_notify_new_fluffy_block(int command, NOTIFY_NEW_FLUFFY_BLOCK::request& arg, cryptonote_connection_context& context)
{
{
MLOGIF_P2P_MESSAGE(crypto::hash hash; cryptonote::block b; bool ret = cryptonote::parse_and_validate_block_from_blob(arg.b.block, b, &hash);, ret, "Received NOTIFY_NEW_FLUFFY_BLOCK " << hash << " (height " << arg.current_blockchain_height << ", " << arg.b.txs.size() << " txes)");
MLOGIF_P2P_MESSAGE(crypto::hash hash; cryptonote::block b; bool ret = cryptonote::parse_and_validate_block_from_blob(arg.b.block, b, &hash);, ret, context << "Received NOTIFY_NEW_FLUFFY_BLOCK " << hash << " (height " << arg.current_blockchain_height << ", " << arg.b.txs.size() << " txes)");
if(context.m_state != cryptonote_connection_context::state_normal)
if(context.m_state != cryptonote_connection_context::state_normal)
return 1;
return 1;
if(!is_synchronized() || m_no_sync ) // can happen if a peer connection goes to normal but another thread still hasn't finished adding queued blocks
if(!is_synchronized()) // can happen if a peer connection goes to normal but another thread still hasn't finished adding queued blocks
{
{
LOG_DEBUG_CC(context, "Received new block while syncing, ignored");
LOG_DEBUG_CC(context, "Received new block while syncing, ignored");
return 1;
return 1;
@ -762,9 +775,12 @@ namespace cryptonote
context.m_needed_objects.clear();
context.m_needed_objects.clear();
context.m_state = cryptonote_connection_context::state_synchronizing;
context.m_state = cryptonote_connection_context::state_synchronizing;
NOTIFY_REQUEST_CHAIN::request r = {};
NOTIFY_REQUEST_CHAIN::request r = {};
context.m_expect_height = m_core.get_current_blockchain_height();
m_core.get_short_chain_history(r.block_ids);
m_core.get_short_chain_history(r.block_ids);
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
r.prune = m_sync_pruned_blocks;
r.prune = m_sync_pruned_blocks;
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
context.m_expect_response = NOTIFY_RESPONSE_CHAIN_ENTRY::ID;
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() );
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() );
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
MLOG_PEER_STATE("requesting chain");
MLOG_PEER_STATE("requesting chain");
@ -929,7 +945,7 @@ namespace cryptonote
// while syncing, core will lock for a long time, so we ignore
// while syncing, core will lock for a long time, so we ignore
// those txes as they aren't really needed anyway, and avoid a
// those txes as they aren't really needed anyway, and avoid a
// long block before replying
// long block before replying
if(!is_synchronized() || m_no_sync )
if(!is_synchronized())
{
{
LOG_DEBUG_CC(context, "Received new tx while syncing, ignored");
LOG_DEBUG_CC(context, "Received new tx while syncing, ignored");
return 1;
return 1;
@ -1029,6 +1045,7 @@ namespace cryptonote
drop_connection(context, false, false);
drop_connection(context, false, false);
return 1;
return 1;
}
}
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
MLOG_P2P_MESSAGE("-->>NOTIFY_RESPONSE_GET_OBJECTS: blocks.size()="
MLOG_P2P_MESSAGE("-->>NOTIFY_RESPONSE_GET_OBJECTS: blocks.size()="
<< rsp.blocks.size() << ", rsp.m_current_blockchain_height=" << rsp.current_blockchain_height
<< rsp.blocks.size() << ", rsp.m_current_blockchain_height=" << rsp.current_blockchain_height
<< ", missed_ids.size()=" << rsp.missed_ids.size());
<< ", missed_ids.size()=" << rsp.missed_ids.size());
@ -1062,6 +1079,14 @@ namespace cryptonote
boost::posix_time::ptime request_time = context.m_last_request_time;
boost::posix_time::ptime request_time = context.m_last_request_time;
context.m_last_request_time = boost::date_time::not_a_date_time;
context.m_last_request_time = boost::date_time::not_a_date_time;
if (context.m_expect_response != NOTIFY_RESPONSE_GET_OBJECTS::ID)
{
LOG_ERROR_CCONTEXT("Got NOTIFY_RESPONSE_GET_OBJECTS out of the blue, dropping connection");
drop_connection(context, true, false);
return 1;
}
context.m_expect_response = 0;
// calculate size of request
// calculate size of request
size_t size = 0;
size_t size = 0;
size_t blocks_size = 0;
size_t blocks_size = 0;
@ -1107,6 +1132,11 @@ namespace cryptonote
return 1;
return 1;
}
}
if (arg.current_blockchain_height < context.m_remote_blockchain_height)
{
MINFO(context << "Claims " << arg.current_blockchain_height << ", claimed " << context.m_remote_blockchain_height << " before");
hit_score(context, 1);
}
context.m_remote_blockchain_height = arg.current_blockchain_height;
context.m_remote_blockchain_height = arg.current_blockchain_height;
if (context.m_remote_blockchain_height > m_core.get_target_blockchain_height())
if (context.m_remote_blockchain_height > m_core.get_target_blockchain_height())
m_core.set_target_blockchain_height(context.m_remote_blockchain_height);
m_core.set_target_blockchain_height(context.m_remote_blockchain_height);
@ -1141,7 +1171,16 @@ namespace cryptonote
return 1;
return 1;
}
}
if (start_height == std::numeric_limits<uint64_t>::max())
if (start_height == std::numeric_limits<uint64_t>::max())
{
start_height = boost::get<txin_gen>(b.miner_tx.vin[0]).height;
start_height = boost::get<txin_gen>(b.miner_tx.vin[0]).height;
if (start_height > context.m_expect_height)
{
LOG_ERROR_CCONTEXT("sent block ahead of expected height, dropping connection");
drop_connection(context, false, false);
++m_sync_bad_spans_downloaded;
return 1;
}
}
auto req_it = context.m_requested_objects.find(block_hash);
auto req_it = context.m_requested_objects.find(block_hash);
if(req_it == context.m_requested_objects.end())
if(req_it == context.m_requested_objects.end())
@ -1230,7 +1269,7 @@ namespace cryptonote
const boost::posix_time::time_duration dt = now - request_time;
const boost::posix_time::time_duration dt = now - request_time;
const float rate = size * 1e6 / (dt.total_microseconds() + 1);
const float rate = size * 1e6 / (dt.total_microseconds() + 1);
MDEBUG(context << " adding span: " << arg.blocks.size() << " at height " << start_height << ", " << dt.total_microseconds()/1e6 << " seconds, " << (rate/1024) << " kB/s, size now " << (m_block_queue.get_data_size() + blocks_size) / 1048576.f << " MB");
MDEBUG(context << " adding span: " << arg.blocks.size() << " at height " << start_height << ", " << dt.total_microseconds()/1e6 << " seconds, " << (rate/1024) << " kB/s, size now " << (m_block_queue.get_data_size() + blocks_size) / 1048576.f << " MB");
m_block_queue.add_blocks(start_height, arg.blocks, context.m_connection_id, rate, blocks_size);
m_block_queue.add_blocks(start_height, arg.blocks, context.m_connection_id, context.m_remote_address, rate, blocks_size);
const crypto::hash last_block_hash = cryptonote::get_block_hash(b);
const crypto::hash last_block_hash = cryptonote::get_block_hash(b);
context.m_last_known_hash = last_block_hash;
context.m_last_known_hash = last_block_hash;
@ -1331,7 +1370,8 @@ namespace cryptonote
uint64_t start_height;
uint64_t start_height;
std::vector<cryptonote::block_complete_entry> blocks;
std::vector<cryptonote::block_complete_entry> blocks;
boost::uuids::uuid span_connection_id;
boost::uuids::uuid span_connection_id;
if (!m_block_queue.get_next_span(start_height, blocks, span_connection_id))
epee::net_utils::network_address span_origin;
if (!m_block_queue.get_next_span(start_height, blocks, span_connection_id, span_origin))
{
{
MDEBUG(context << " no next span found, going back to download");
MDEBUG(context << " no next span found, going back to download");
break;
break;
@ -1429,6 +1469,7 @@ namespace cryptonote
if (!m_core.prepare_handle_incoming_blocks(blocks, pblocks))
if (!m_core.prepare_handle_incoming_blocks(blocks, pblocks))
{
{
LOG_ERROR_CCONTEXT("Failure in prepare_handle_incoming_blocks");
LOG_ERROR_CCONTEXT("Failure in prepare_handle_incoming_blocks");
drop_connections(span_origin);
return 1;
return 1;
}
}
if (!pblocks.empty() && pblocks.size() != blocks.size())
if (!pblocks.empty() && pblocks.size() != blocks.size())
@ -1468,6 +1509,7 @@ namespace cryptonote
{
{
if(tvc[i].m_verifivation_failed)
if(tvc[i].m_verifivation_failed)
{
{
drop_connections(span_origin);
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
cryptonote::transaction tx;
cryptonote::transaction tx;
crypto::hash txid;
crypto::hash txid;
@ -1509,6 +1551,7 @@ namespace cryptonote
if(bvc.m_verifivation_failed)
if(bvc.m_verifivation_failed)
{
{
drop_connections(span_origin);
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
LOG_PRINT_CCONTEXT_L1("Block verification failed, dropping connection");
LOG_PRINT_CCONTEXT_L1("Block verification failed, dropping connection");
drop_connection_with_score(context, bvc.m_bad_pow ? P2P_IP_FAILS_BEFORE_BLOCK : 1, true);
drop_connection_with_score(context, bvc.m_bad_pow ? P2P_IP_FAILS_BEFORE_BLOCK : 1, true);
@ -1528,6 +1571,7 @@ namespace cryptonote
}
}
if(bvc.m_marked_as_orphaned)
if(bvc.m_marked_as_orphaned)
{
{
drop_connections(span_origin);
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
if (!m_p2p->for_connection(span_connection_id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
LOG_PRINT_CCONTEXT_L1("Block received at sync phase was marked as orphaned, dropping connection");
LOG_PRINT_CCONTEXT_L1("Block received at sync phase was marked as orphaned, dropping connection");
drop_connection(context, true, true);
drop_connection(context, true, true);
@ -1666,24 +1710,46 @@ skip:
bool t_cryptonote_protocol_handler<t_core>::kick_idle_peers()
bool t_cryptonote_protocol_handler<t_core>::kick_idle_peers()
{
{
MTRACE("Checking for idle peers...");
MTRACE("Checking for idle peers...");
std::vector<std::pair<boost::uuids::uuid, unsigned>> idle_peers;
m_p2p->for_each_connection([&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t support_flags)->bool
m_p2p->for_each_connection([&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t support_flags)->bool
{
{
if (context.m_state == cryptonote_connection_context::state_synchronizing && context.m_last_request_time != boost::date_time::not_a_date_time)
if (context.m_state == cryptonote_connection_context::state_synchronizing && context.m_last_request_time != boost::date_time::not_a_date_time)
{
{
const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time();
const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time();
const boost::posix_time::time_duration dt = now - context.m_last_request_time;
const boost::posix_time::time_duration dt = now - context.m_last_request_time;
if (dt.total_microseconds() > IDLE_PEER_KICK_TIME)
const auto ms = dt.total_microseconds();
if (ms > IDLE_PEER_KICK_TIME || (context.m_expect_response && ms > NON_RESPONSIVE_PEER_KICK_TIME))
{
{
MINFO(context << " kicking idle peer, last update " << (dt.total_microseconds() / 1.e6) << " seconds ago");
if (context.m_score-- >= 0)
LOG_PRINT_CCONTEXT_L2("requesting callback");
{
context.m_last_request_time = boost::date_time::not_a_date_time;
MINFO(context << " kicking idle peer, last update " << (dt.total_microseconds() / 1.e6) << " seconds ago, expecting " << (int)context.m_expect_response);
context.m_state = cryptonote_connection_context::state_standby; // we'll go back to adding, then (if we can't), download
LOG_PRINT_CCONTEXT_L2("requesting callback");
++context.m_callback_request_count;
context.m_last_request_time = boost::date_time::not_a_date_time;
m_p2p->request_callback(context);
context.m_expect_response = 0;
context.m_expect_height = 0;
context.m_state = cryptonote_connection_context::state_standby; // we'll go back to adding, then (if we can't), download
++context.m_callback_request_count;
m_p2p->request_callback(context);
}
else
{
idle_peers.push_back(std::make_pair(context.m_connection_id, context.m_expect_response == 0 ? 1 : 5));
}
}
}
}
}
return true;
return true;
});
});
for (const auto &e: idle_peers)
{
const auto &uuid = e.first;
m_p2p->for_connection(uuid, [&](cryptonote_connection_context& ctx, nodetool::peerid_type peer_id, uint32_t f)->bool{
MINFO(ctx << "dropping idle peer with negative score");
drop_connection_with_score(ctx, e.second, false);
return true;
});
}
return true;
return true;
}
}
//------------------------------------------------------------------------------------------------------------------------
//------------------------------------------------------------------------------------------------------------------------
@ -1760,6 +1826,16 @@ skip:
LOG_ERROR_CCONTEXT("Failed to handle NOTIFY_REQUEST_CHAIN.");
LOG_ERROR_CCONTEXT("Failed to handle NOTIFY_REQUEST_CHAIN.");
return 1;
return 1;
}
}
if (r.m_block_ids.size() >= 2)
{
cryptonote::block b;
if (!m_core.get_block_by_hash(r.m_block_ids[1], b))
{
LOG_ERROR_CCONTEXT("Failed to handle NOTIFY_REQUEST_CHAIN: first block not found");
return 1;
}
r.first_block = cryptonote::block_to_blob(b);
}
MLOG_P2P_MESSAGE("-->>NOTIFY_RESPONSE_CHAIN_ENTRY: m_start_height=" << r.start_height << ", m_total_height=" << r.total_height << ", m_block_ids.size()=" << r.m_block_ids.size());
MLOG_P2P_MESSAGE("-->>NOTIFY_RESPONSE_CHAIN_ENTRY: m_start_height=" << r.start_height << ", m_total_height=" << r.total_height << ", m_block_ids.size()=" << r.m_block_ids.size());
post_notify<NOTIFY_RESPONSE_CHAIN_ENTRY>(r, context);
post_notify<NOTIFY_RESPONSE_CHAIN_ENTRY>(r, context);
return 1;
return 1;
@ -1944,7 +2020,7 @@ skip:
if (local_stripe == 0)
if (local_stripe == 0)
return false;
return false;
// don't request pre-bulletprooof pruned blocks, we can't reconstruct their weight (yet)
// don't request pre-bulletprooof pruned blocks, we can't reconstruct their weight (yet)
static const uint64_t bp_fork_height = m_core.get_earliest_ideal_height_for_version(HF_VERSION_SMALLER_BP);
static const uint64_t bp_fork_height = m_core.get_earliest_ideal_height_for_version(HF_VERSION_SMALLER_BP + 1 );
if (first_block_height < bp_fork_height)
if (first_block_height < bp_fork_height)
return false;
return false;
// assumes the span size is less or equal to the stripe size
// assumes the span size is less or equal to the stripe size
@ -2125,7 +2201,7 @@ skip:
const uint64_t first_block_height = context.m_last_response_height - context.m_needed_objects.size() + 1;
const uint64_t first_block_height = context.m_last_response_height - context.m_needed_objects.size() + 1;
static const uint64_t bp_fork_height = m_core.get_earliest_ideal_height_for_version(HF_VERSION_SMALLER_BP);
static const uint64_t bp_fork_height = m_core.get_earliest_ideal_height_for_version(HF_VERSION_SMALLER_BP);
bool sync_pruned_blocks = m_sync_pruned_blocks && first_block_height >= bp_fork_height && m_core.get_blockchain_pruning_seed();
bool sync_pruned_blocks = m_sync_pruned_blocks && first_block_height >= bp_fork_height && m_core.get_blockchain_pruning_seed();
span = m_block_queue.reserve_span(first_block_height, context.m_last_response_height, count_limit, context.m_connection_id, sync_pruned_blocks, m_core.get_blockchain_pruning_seed(), context.m_pruning_seed, context.m_remote_blockchain_height, context.m_needed_objects);
span = m_block_queue.reserve_span(first_block_height, context.m_last_response_height, count_limit, context.m_connection_id, context.m_remote_address, sync_pruned_blocks, m_core.get_blockchain_pruning_seed(), context.m_pruning_seed, context.m_remote_blockchain_height, context.m_needed_objects);
MDEBUG(context << " span from " << first_block_height << ": " << span.first << "/" << span.second);
MDEBUG(context << " span from " << first_block_height << ": " << span.first << "/" << span.second);
if (span.second > 0)
if (span.second > 0)
{
{
@ -2206,6 +2282,8 @@ skip:
}
}
}
}
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
context.m_expect_height = span.first;
context.m_expect_response = NOTIFY_RESPONSE_GET_OBJECTS::ID;
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_GET_OBJECTS: blocks.size()=" << req.blocks.size()
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_GET_OBJECTS: blocks.size()=" << req.blocks.size()
<< "requested blocks count=" << count << " / " << count_limit << " from " << span.first << ", first hash " << req.blocks.front());
<< "requested blocks count=" << count << " / " << count_limit << " from " << span.first << ", first hash " << req.blocks.front());
//epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
//epee::net_utils::network_throttle_manager::get_global_throttle_inreq().logger_handle_net("log/dr-monero/net/req-all.data", sec, get_avg_block_size());
@ -2260,8 +2338,8 @@ skip:
uint64_t start_height;
uint64_t start_height;
std::vector<cryptonote::block_complete_entry> blocks;
std::vector<cryptonote::block_complete_entry> blocks;
boost::uuids::uuid span_connection_id;
boost::uuids::uuid span_connection_id;
bool filled = false ;
epee::net_utils::network_address span_origin ;
if (m_block_queue.get_next_span(start_height, blocks, span_connection_id, filled) && filled )
if (m_block_queue.get_next_span(start_height, blocks, span_connection_id, span_origin, true) )
{
{
LOG_DEBUG_CC(context, "No other thread is adding blocks, resuming");
LOG_DEBUG_CC(context, "No other thread is adding blocks, resuming");
MLOG_PEER_STATE("will try to add blocks next");
MLOG_PEER_STATE("will try to add blocks next");
@ -2277,6 +2355,7 @@ skip:
{//we have to fetch more objects ids, request blockchain entry
{//we have to fetch more objects ids, request blockchain entry
NOTIFY_REQUEST_CHAIN::request r = {};
NOTIFY_REQUEST_CHAIN::request r = {};
context.m_expect_height = m_core.get_current_blockchain_height();
m_core.get_short_chain_history(r.block_ids);
m_core.get_short_chain_history(r.block_ids);
CHECK_AND_ASSERT_MES(!r.block_ids.empty(), false, "Short chain history is empty");
CHECK_AND_ASSERT_MES(!r.block_ids.empty(), false, "Short chain history is empty");
@ -2284,7 +2363,10 @@ skip:
{
{
// we'll want to start off from where we are on that peer, which may not be added yet
// we'll want to start off from where we are on that peer, which may not be added yet
if (context.m_last_known_hash != crypto::null_hash && r.block_ids.front() != context.m_last_known_hash)
if (context.m_last_known_hash != crypto::null_hash && r.block_ids.front() != context.m_last_known_hash)
{
context.m_expect_height = std::numeric_limits<uint64_t>::max();
r.block_ids.push_front(context.m_last_known_hash);
r.block_ids.push_front(context.m_last_known_hash);
}
}
}
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
handler_request_blocks_history( r.block_ids ); // change the limit(?), sleep(?)
@ -2296,6 +2378,7 @@ skip:
//LOG_PRINT_CCONTEXT_L1("r = " << 200);
//LOG_PRINT_CCONTEXT_L1("r = " << 200);
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
context.m_last_request_time = boost::posix_time::microsec_clock::universal_time();
context.m_expect_response = NOTIFY_RESPONSE_CHAIN_ENTRY::ID;
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() << ", start_from_current_chain " << start_from_current_chain);
MLOG_P2P_MESSAGE("-->>NOTIFY_REQUEST_CHAIN: m_block_ids.size()=" << r.block_ids.size() << ", start_from_current_chain " << start_from_current_chain);
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
post_notify<NOTIFY_REQUEST_CHAIN>(r, context);
MLOG_PEER_STATE("requesting chain");
MLOG_PEER_STATE("requesting chain");
@ -2321,7 +2404,7 @@ skip:
}
}
else
else
{
{
MINFO(context << " we've reached this peer's blockchain height");
MINFO(context << " we've reached this peer's blockchain height (theirs " << context.m_remote_blockchain_height << ", our target " << m_core.get_target_blockchain_height() );
}
}
}
}
return true;
return true;
@ -2418,6 +2501,20 @@ skip:
<< ", m_start_height=" << arg.start_height << ", m_total_height=" << arg.total_height);
<< ", m_start_height=" << arg.start_height << ", m_total_height=" << arg.total_height);
MLOG_PEER_STATE("received chain");
MLOG_PEER_STATE("received chain");
if (context.m_expect_response != NOTIFY_RESPONSE_CHAIN_ENTRY::ID)
{
LOG_ERROR_CCONTEXT("Got NOTIFY_RESPONSE_CHAIN_ENTRY out of the blue, dropping connection");
drop_connection(context, true, false);
return 1;
}
context.m_expect_response = 0;
if (arg.start_height + 1 > context.m_expect_height) // we expect an overlapping block
{
LOG_ERROR_CCONTEXT("Got NOTIFY_RESPONSE_CHAIN_ENTRY past expected height, dropping connection");
drop_connection(context, true, false);
return 1;
}
context.m_last_request_time = boost::date_time::not_a_date_time;
context.m_last_request_time = boost::date_time::not_a_date_time;
m_sync_download_chain_size += arg.m_block_ids.size() * sizeof(crypto::hash);
m_sync_download_chain_size += arg.m_block_ids.size() * sizeof(crypto::hash);
@ -2442,12 +2539,17 @@ skip:
}
}
MDEBUG(context << "first block hash " << arg.m_block_ids.front() << ", last " << arg.m_block_ids.back());
MDEBUG(context << "first block hash " << arg.m_block_ids.front() << ", last " << arg.m_block_ids.back());
if (arg.total_height >= CRYPTONOTE_MAX_BLOCK_NUMBER || arg.m_block_ids.size() >= CRYPTONOTE_MAX_BLOCK_NUMBER )
if (arg.total_height >= CRYPTONOTE_MAX_BLOCK_NUMBER || arg.m_block_ids.size() > BLOCKS_IDS_SYNCHRONIZING_MAX_COUNT )
{
{
LOG_ERROR_CCONTEXT("sent wrong NOTIFY_RESPONSE_CHAIN_ENTRY, with total_height=" << arg.total_height << " and block_ids=" << arg.m_block_ids.size());
LOG_ERROR_CCONTEXT("sent wrong NOTIFY_RESPONSE_CHAIN_ENTRY, with total_height=" << arg.total_height << " and block_ids=" << arg.m_block_ids.size());
drop_connection(context, false, false);
drop_connection(context, false, false);
return 1;
return 1;
}
}
if (arg.total_height < context.m_remote_blockchain_height)
{
MINFO(context << "Claims " << arg.total_height << ", claimed " << context.m_remote_blockchain_height << " before");
hit_score(context, 1);
}
context.m_remote_blockchain_height = arg.total_height;
context.m_remote_blockchain_height = arg.total_height;
context.m_last_response_height = arg.start_height + arg.m_block_ids.size()-1;
context.m_last_response_height = arg.start_height + arg.m_block_ids.size()-1;
if(context.m_last_response_height > context.m_remote_blockchain_height)
if(context.m_last_response_height > context.m_remote_blockchain_height)
@ -2459,8 +2561,19 @@ skip:
return 1;
return 1;
}
}
std::unordered_set<crypto::hash> hashes;
for (const auto &h: arg.m_block_ids)
{
if (!hashes.insert(h).second)
{
LOG_ERROR_CCONTEXT("sent duplicate block, dropping connection");
drop_connection(context, true, false);
return 1;
}
}
uint64_t n_use_blocks = m_core.prevalidate_block_hashes(arg.start_height, arg.m_block_ids, arg.m_block_weights);
uint64_t n_use_blocks = m_core.prevalidate_block_hashes(arg.start_height, arg.m_block_ids, arg.m_block_weights);
if (n_use_blocks + HASH_OF_HASHES_STEP <= arg.m_block_ids.size())
if (n_use_blocks == 0 || n_use_blocks + HASH_OF_HASHES_STEP <= arg.m_block_ids.size())
{
{
LOG_ERROR_CCONTEXT("Most blocks are invalid, dropping connection");
LOG_ERROR_CCONTEXT("Most blocks are invalid, dropping connection");
drop_connection(context, true, false);
drop_connection(context, true, false);
@ -2469,8 +2582,15 @@ skip:
context.m_needed_objects.clear();
context.m_needed_objects.clear();
uint64_t added = 0;
uint64_t added = 0;
std::unordered_set<crypto::hash> blocks_found;
for (size_t i = 0; i < arg.m_block_ids.size(); ++i)
for (size_t i = 0; i < arg.m_block_ids.size(); ++i)
{
{
if (!blocks_found.insert(arg.m_block_ids[i]).second)
{
LOG_ERROR_CCONTEXT("Duplicate blocks in chain entry response, dropping connection");
drop_connection(context, true, false);
return 1;
}
const uint64_t block_weight = arg.m_block_weights.empty() ? 0 : arg.m_block_weights[i];
const uint64_t block_weight = arg.m_block_weights.empty() ? 0 : arg.m_block_weights[i];
context.m_needed_objects.push_back(std::make_pair(arg.m_block_ids[i], block_weight));
context.m_needed_objects.push_back(std::make_pair(arg.m_block_ids[i], block_weight));
if (++added == n_use_blocks)
if (++added == n_use_blocks)
@ -2523,15 +2643,15 @@ skip:
// send fluffy ones first, we want to encourage people to run that
// send fluffy ones first, we want to encourage people to run that
if (!fluffyConnections.empty())
if (!fluffyConnections.empty())
{
{
std::string fluffyBlob;
epee::byte_slice fluffyBlob;
epee::serialization::store_t_to_binary(fluffy_arg, fluffyBlob);
epee::serialization::store_t_to_binary(fluffy_arg, fluffyBlob, 32 * 1024 );
m_p2p->relay_notify_to_list(NOTIFY_NEW_FLUFFY_BLOCK::ID, epee::strspan<uint8_t> (fluffyBlob), std::move(fluffyConnections));
m_p2p->relay_notify_to_list(NOTIFY_NEW_FLUFFY_BLOCK::ID, epee::to_span (fluffyBlob), std::move(fluffyConnections));
}
}
if (!fullConnections.empty())
if (!fullConnections.empty())
{
{
std::string fullBlob;
epee::byte_slice fullBlob;
epee::serialization::store_t_to_binary(arg, fullBlob);
epee::serialization::store_t_to_binary(arg, fullBlob, 128 * 1024 );
m_p2p->relay_notify_to_list(NOTIFY_NEW_BLOCK::ID, epee::strspan<uint8_t> (fullBlob), std::move(fullConnections));
m_p2p->relay_notify_to_list(NOTIFY_NEW_BLOCK::ID, epee::to_span (fullBlob), std::move(fullConnections));
}
}
return true;
return true;
@ -2564,6 +2684,19 @@ skip:
}
}
//------------------------------------------------------------------------------------------------------------------------
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
template<class t_core>
void t_cryptonote_protocol_handler<t_core>::hit_score(cryptonote_connection_context &context, int32_t score)
{
if (score <= 0)
{
MERROR("Negative score hit");
return;
}
context.m_score -= score;
if (context.m_score <= DROP_PEERS_ON_SCORE)
drop_connection_with_score(context, 5, false);
}
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
std::string t_cryptonote_protocol_handler<t_core>::get_peers_overview() const
std::string t_cryptonote_protocol_handler<t_core>::get_peers_overview() const
{
{
std::stringstream ss;
std::stringstream ss;
@ -2638,6 +2771,13 @@ skip:
}
}
//------------------------------------------------------------------------------------------------------------------------
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
template<class t_core>
bool t_cryptonote_protocol_handler<t_core>::is_busy_syncing()
{
const boost::unique_lock<boost::mutex> sync{m_sync_lock, boost::try_to_lock};
return !sync.owns_lock();
}
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
void t_cryptonote_protocol_handler<t_core>::drop_connection_with_score(cryptonote_connection_context &context, unsigned score, bool flush_all_spans)
void t_cryptonote_protocol_handler<t_core>::drop_connection_with_score(cryptonote_connection_context &context, unsigned score, bool flush_all_spans)
{
{
LOG_DEBUG_CC(context, "dropping connection id " << context.m_connection_id << " (pruning seed " <<
LOG_DEBUG_CC(context, "dropping connection id " << context.m_connection_id << " (pruning seed " <<
@ -2659,6 +2799,29 @@ skip:
}
}
//------------------------------------------------------------------------------------------------------------------------
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
template<class t_core>
void t_cryptonote_protocol_handler<t_core>::drop_connections(const epee::net_utils::network_address address)
{
MWARNING("dropping connections to " << address.str());
m_p2p->add_host_fail(address, 5);
std::vector<boost::uuids::uuid> drop;
m_p2p->for_each_connection([&](const connection_context& cntxt, nodetool::peerid_type peer_id, uint32_t support_flags) {
if (address.is_same_host(cntxt.m_remote_address))
drop.push_back(cntxt.m_connection_id);
return true;
});
for (const boost::uuids::uuid &id: drop)
{
m_block_queue.flush_spans(id, true);
m_p2p->for_connection(id, [&](cryptonote_connection_context& context, nodetool::peerid_type peer_id, uint32_t f)->bool{
drop_connection(context, true, false);
return true;
});
}
}
//------------------------------------------------------------------------------------------------------------------------
template<class t_core>
void t_cryptonote_protocol_handler<t_core>::on_connection_close(cryptonote_connection_context &context)
void t_cryptonote_protocol_handler<t_core>::on_connection_close(cryptonote_connection_context &context)
{
{
uint64_t target = 0;
uint64_t target = 0;
