Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Vote replay in crawler #2497

Merged
merged 3 commits into from
Jan 25, 2020
+42 −69
Merged

Vote replay in crawler #2497

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e7a17c6
Move vote replay code in to rep_crawler since this is a very infreque…
clemahieu Jan 20, 2020
48aef31
Directly calling vote_blocking instead of creating a new deque of val…
clemahieu Jan 20, 2020
db75908
Fixing timeout for possibly slow test situations.
clemahieu Jan 24, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 9 additions & 10 deletions nano/core_test/ledger.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -742,11 +742,10 @@ TEST (votes, check_signature)
}
auto vote1 (std::make_shared<nano::vote> (nano::test_genesis_key.pub, nano::test_genesis_key.prv, 1, send1));
vote1->signature.bytes[0] ^= 1;
auto transaction (node1.store.tx_begin_read ());
ASSERT_EQ (nano::vote_code::invalid, node1.vote_processor.vote_blocking (transaction, vote1, std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, nano::endpoint (boost::asio::ip::address_v6 (), 0), node1.network_params.protocol.protocol_version)));
ASSERT_EQ (nano::vote_code::invalid, node1.vote_processor.vote_blocking (vote1, std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, nano::endpoint (boost::asio::ip::address_v6 (), 0), node1.network_params.protocol.protocol_version)));
vote1->signature.bytes[0] ^= 1;
ASSERT_EQ (nano::vote_code::vote, node1.vote_processor.vote_blocking (transaction, vote1, std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, nano::endpoint (boost::asio::ip::address_v6 (), 0), node1.network_params.protocol.protocol_version)));
ASSERT_EQ (nano::vote_code::replay, node1.vote_processor.vote_blocking (transaction, vote1, std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, nano::endpoint (boost::asio::ip::address_v6 (), 0), node1.network_params.protocol.protocol_version)));
ASSERT_EQ (nano::vote_code::vote, node1.vote_processor.vote_blocking (vote1, std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, nano::endpoint (boost::asio::ip::address_v6 (), 0), node1.network_params.protocol.protocol_version)));
ASSERT_EQ (nano::vote_code::replay, node1.vote_processor.vote_blocking (vote1, std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, nano::endpoint (boost::asio::ip::address_v6 (), 0), node1.network_params.protocol.protocol_version)));
}

TEST (votes, add_one)
Expand Down Expand Up @@ -878,13 +877,13 @@ TEST (votes, add_old)
votes1 = node1.active.roots.find (send1->qualified_root ())->election;
}
auto channel (std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, node1.network.endpoint (), node1.network_params.protocol.protocol_version));
node1.vote_processor.vote_blocking (transaction, vote1, channel);
node1.vote_processor.vote_blocking (vote1, channel);
nano::keypair key2;
auto send2 (std::make_shared<nano::send_block> (genesis.hash (), key2.pub, 0, nano::test_genesis_key.prv, nano::test_genesis_key.pub, 0));
node1.work_generate_blocking (*send2);
auto vote2 (std::make_shared<nano::vote> (nano::test_genesis_key.pub, nano::test_genesis_key.prv, 1, send2));
votes1->last_votes[nano::test_genesis_key.pub].time = std::chrono::steady_clock::now () - std::chrono::seconds (20);
node1.vote_processor.vote_blocking (transaction, vote2, channel);
node1.vote_processor.vote_blocking (vote2, channel);
ASSERT_EQ (2, votes1->last_votes_size ());
ASSERT_NE (votes1->last_votes.end (), votes1->last_votes.find (nano::test_genesis_key.pub));
ASSERT_EQ (send1->hash (), votes1->last_votes[nano::test_genesis_key.pub].hash);
Expand Down Expand Up @@ -919,12 +918,12 @@ TEST (votes, add_old_different_account)
ASSERT_EQ (1, votes2->last_votes_size ());
auto vote1 (std::make_shared<nano::vote> (nano::test_genesis_key.pub, nano::test_genesis_key.prv, 2, send1));
auto channel (std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, node1.network.endpoint (), node1.network_params.protocol.protocol_version));
auto vote_result1 (node1.vote_processor.vote_blocking (transaction, vote1, channel));
auto vote_result1 (node1.vote_processor.vote_blocking (vote1, channel));
ASSERT_EQ (nano::vote_code::vote, vote_result1);
ASSERT_EQ (2, votes1->last_votes.size ());
ASSERT_EQ (1, votes2->last_votes.size ());
auto vote2 (std::make_shared<nano::vote> (nano::test_genesis_key.pub, nano::test_genesis_key.prv, 1, send2));
auto vote_result2 (node1.vote_processor.vote_blocking (transaction, vote2, channel));
auto vote_result2 (node1.vote_processor.vote_blocking (vote2, channel));
ASSERT_EQ (nano::vote_code::vote, vote_result2);
ASSERT_EQ (2, votes1->last_votes.size ());
ASSERT_EQ (2, votes2->last_votes.size ());
Expand Down Expand Up @@ -957,12 +956,12 @@ TEST (votes, add_cooldown)
}
auto vote1 (std::make_shared<nano::vote> (nano::test_genesis_key.pub, nano::test_genesis_key.prv, 1, send1));
auto channel (std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, node1.network.endpoint (), node1.network_params.protocol.protocol_version));
node1.vote_processor.vote_blocking (transaction, vote1, channel);
node1.vote_processor.vote_blocking (vote1, channel);
nano::keypair key2;
auto send2 (std::make_shared<nano::send_block> (genesis.hash (), key2.pub, 0, nano::test_genesis_key.prv, nano::test_genesis_key.pub, 0));
node1.work_generate_blocking (*send2);
auto vote2 (std::make_shared<nano::vote> (nano::test_genesis_key.pub, nano::test_genesis_key.prv, 2, send2));
node1.vote_processor.vote_blocking (transaction, vote2, channel);
node1.vote_processor.vote_blocking (vote2, channel);
ASSERT_EQ (2, votes1->last_votes.size ());
ASSERT_NE (votes1->last_votes.end (), votes1->last_votes.find (nano::test_genesis_key.pub));
ASSERT_EQ (send1->hash (), votes1->last_votes[nano::test_genesis_key.pub].hash);
Expand Down
37 changes: 14 additions & 23 deletions nano/core_test/node.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2221,34 +2221,31 @@ TEST (node, send_callback)
// This helps representatives continue from their last sequence number if their node is reinitialized and the old sequence number is lost
TEST (node, vote_replay)
{
nano::system system (2);
nano::system system (1);
auto & node1 (*system.nodes[0]);
auto & node2 (*system.nodes[1]);
nano::keypair key;
auto open (std::make_shared<nano::open_block> (0, 1, key.pub, key.prv, key.pub, 0));
node1.work_generate_blocking (*open);
nano::genesis genesis;
for (auto i (0); i < 11000; ++i)
{
auto transaction (node2.store.tx_begin_read ());
auto vote (node2.store.vote_generate (transaction, nano::test_genesis_key.pub, nano::test_genesis_key.prv, open));
auto transaction (node1.store.tx_begin_read ());
auto vote (node1.store.vote_generate (transaction, nano::test_genesis_key.pub, nano::test_genesis_key.prv, genesis.open));
}
auto node2 = system.add_node ();
{
auto transaction (node1.store.tx_begin_read ());
nano::lock_guard<std::mutex> lock (node1.store.get_cache_mutex ());
auto vote (node1.store.vote_current (transaction, nano::test_genesis_key.pub));
auto transaction (node2->store.tx_begin_read ());
nano::lock_guard<std::mutex> lock (node2->store.get_cache_mutex ());
auto vote (node2->store.vote_current (transaction, nano::test_genesis_key.pub));
ASSERT_EQ (nullptr, vote);
}
system.wallet (0)->insert_adhoc (nano::test_genesis_key.prv);
auto block (system.wallet (0)->send_action (nano::test_genesis_key.pub, key.pub, nano::Gxrb_ratio));
ASSERT_NE (nullptr, block);
system.wallet (1)->insert_adhoc (nano::test_genesis_key.prv);
auto done (false);
system.deadline_set (20s);
while (!done)
{
auto ec = system.poll ();
auto transaction (node1.store.tx_begin_read ());
nano::lock_guard<std::mutex> lock (node1.store.get_cache_mutex ());
auto vote (node1.store.vote_current (transaction, nano::test_genesis_key.pub));
auto transaction (node2->store.tx_begin_read ());
nano::lock_guard<std::mutex> lock (node2->store.get_cache_mutex ());
auto vote (node2->store.vote_current (transaction, nano::test_genesis_key.pub));
done = vote && (vote->sequence >= 10000);
ASSERT_NO_ERROR (ec);
}
Expand Down Expand Up @@ -2925,10 +2922,7 @@ TEST (node, fork_invalid_block_signature_vote_by_hash)
std::vector<nano::block_hash> vote_blocks;
vote_blocks.push_back (send2->hash ());
auto vote (std::make_shared<nano::vote> (nano::test_genesis_key.pub, nano::test_genesis_key.prv, 0, vote_blocks));
{
auto transaction (node1.store.tx_begin_read ());
node1.vote_processor.vote_blocking (transaction, vote, std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, node1.network.endpoint (), node1.network_params.protocol.protocol_version));
}
node1.vote_processor.vote_blocking (vote, std::make_shared<nano::transport::channel_udp> (node1.network.udp_channels, node1.network.endpoint (), node1.network_params.protocol.protocol_version));
while (node1.block (send1->hash ()))
{
ASSERT_NO_ERROR (system.poll ());
Expand Down Expand Up @@ -3094,10 +3088,7 @@ TEST (node, confirm_back)
std::vector<nano::block_hash> vote_blocks;
vote_blocks.push_back (send2->hash ());
auto vote (std::make_shared<nano::vote> (nano::test_genesis_key.pub, nano::test_genesis_key.prv, 0, vote_blocks));
{
auto transaction (node.store.tx_begin_read ());
node.vote_processor.vote_blocking (transaction, vote, std::make_shared<nano::transport::channel_udp> (node.network.udp_channels, node.network.endpoint (), node.network_params.protocol.protocol_version));
}
node.vote_processor.vote_blocking (vote, std::make_shared<nano::transport::channel_udp> (node.network.udp_channels, node.network.endpoint (), node.network_params.protocol.protocol_version));
system.deadline_set (10s);
while (!node.active.empty ())
{
Expand Down
2 changes: 1 addition & 1 deletion nano/node/node.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ bootstrap_initiator (*this),
bootstrap (config.peering_port, *this),
application_path (application_path_a),
port_mapping (*this),
vote_processor (checker, active, store, observers, stats, config, logger, online_reps, ledger, network_params),
vote_processor (checker, active, observers, stats, config, logger, online_reps, ledger, network_params),
rep_crawler (*this),
warmed_up (0),
block_processor (*this, write_database_queue),
Expand Down
10 changes: 10 additions & 0 deletions nano/node/repcrawler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ void nano::rep_crawler::validate ()
nano::lock_guard<std::mutex> lock (active_mutex);
responses_l.swap (responses);
}
auto transaction (node.store.tx_begin_read ());
auto minimum = node.minimum_principal_weight ();
for (auto const & i : responses_l)
{
Expand Down Expand Up @@ -85,6 +86,15 @@ void nano::rep_crawler::validate ()
}
}
}
// This tries to assist rep nodes that have lost track of their highest sequence number by replaying our highest known vote back to them
// Only do this if the sequence number is significantly different to account for network reordering
// Amplify attack considerations: We're sending out a confirm_ack in response to a confirm_ack for no net traffic increase
auto max_vote (node.store.vote_max (transaction, vote));
if (max_vote->sequence > vote->sequence + 10000)
{
nano::confirm_ack confirm (max_vote);
channel->send (confirm); // this is non essential traffic as it will be resolicited if not received
}
}
}

Expand Down
36 changes: 5 additions & 31 deletions nano/node/vote_processor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,9 @@

#include <boost/format.hpp>

nano::vote_processor::vote_processor (nano::signature_checker & checker_a, nano::active_transactions & active_a, nano::block_store & store_a, nano::node_observers & observers_a, nano::stat & stats_a, nano::node_config & config_a, nano::logger_mt & logger_a, nano::online_reps & online_reps_a, nano::ledger & ledger_a, nano::network_params & network_params_a) :
nano::vote_processor::vote_processor (nano::signature_checker & checker_a, nano::active_transactions & active_a, nano::node_observers & observers_a, nano::stat & stats_a, nano::node_config & config_a, nano::logger_mt & logger_a, nano::online_reps & online_reps_a, nano::ledger & ledger_a, nano::network_params & network_params_a) :
checker (checker_a),
active (active_a),
store (store_a),
observers (observers_a),
stats (stats_a),
config (config_a),
Expand Down Expand Up @@ -54,7 +53,7 @@ void nano::vote_processor::process_loop ()
{
if (!votes.empty ())
{
std::deque<std::pair<std::shared_ptr<nano::vote>, std::shared_ptr<nano::transport::channel>>> votes_l;
decltype (votes) votes_l;
votes_l.swap (votes);

log_this_iteration = false;
Expand All @@ -70,20 +69,6 @@ void nano::vote_processor::process_loop ()
is_active = true;
lock.unlock ();
verify_votes (votes_l);
{
auto transaction (store.tx_begin_read ());
uint64_t count (1);
for (auto & i : votes_l)
{
vote_blocking (transaction, i.first, i.second, true);
// Free active_transactions mutex each 100 processed votes
if (count % 100 == 0)
{
transaction.refresh ();
}
count++;
}
}
lock.lock ();
is_active = false;

Expand Down Expand Up @@ -155,7 +140,7 @@ void nano::vote_processor::vote (std::shared_ptr<nano::vote> vote_a, std::shared
}
}

void nano::vote_processor::verify_votes (std::deque<std::pair<std::shared_ptr<nano::vote>, std::shared_ptr<nano::transport::channel>>> & votes_a)
void nano::vote_processor::verify_votes (decltype (votes) const & votes_a)
{
auto size (votes_a.size ());
std::vector<unsigned char const *> messages;
Expand All @@ -178,37 +163,26 @@ void nano::vote_processor::verify_votes (std::deque<std::pair<std::shared_ptr<na
}
nano::signature_check_set check = { size, messages.data (), lengths.data (), pub_keys.data (), signatures.data (), verifications.data () };
checker.verify (check);
std::remove_reference_t<decltype (votes_a)> result;
auto i (0);
for (auto & vote : votes_a)
{
assert (verifications[i] == 1 || verifications[i] == 0);
if (verifications[i] == 1)
{
result.push_back (vote);
vote_blocking (vote.first, vote.second, true);
}
++i;
}
votes_a.swap (result);
}

// node.active.mutex lock required
nano::vote_code nano::vote_processor::vote_blocking (nano::transaction const & transaction_a, std::shared_ptr<nano::vote> vote_a, std::shared_ptr<nano::transport::channel> channel_a, bool validated)
nano::vote_code nano::vote_processor::vote_blocking (std::shared_ptr<nano::vote> vote_a, std::shared_ptr<nano::transport::channel> channel_a, bool validated)
{
auto result (nano::vote_code::invalid);
if (validated || !vote_a->validate ())
{
result = active.vote (vote_a);
observers.vote.notify (vote_a, channel_a, result);
// This tries to assist rep nodes that have lost track of their highest sequence number by replaying our highest known vote back to them
// Only do this if the sequence number is significantly different to account for network reordering
// Amplify attack considerations: We're sending out a confirm_ack in response to a confirm_ack for no net traffic increase
auto max_vote (store.vote_max (transaction_a, vote_a));
if (max_vote->sequence > vote_a->sequence + 10000)
{
nano::confirm_ack confirm (max_vote);
channel_a->send (confirm); // this is non essential traffic as it will be resolicited if not received
}
}
std::string status;
switch (result)
Expand Down
7 changes: 3 additions & 4 deletions nano/node/vote_processor.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,11 @@ namespace transport
class vote_processor final
{
public:
explicit vote_processor (nano::signature_checker & checker_a, nano::active_transactions & active_a, nano::block_store & store_a, nano::node_observers & observers_a, nano::stat & stats_a, nano::node_config & config_a, nano::logger_mt & logger_a, nano::online_reps & online_reps_a, nano::ledger & ledger_a, nano::network_params & network_params_a);
explicit vote_processor (nano::signature_checker & checker_a, nano::active_transactions & active_a, nano::node_observers & observers_a, nano::stat & stats_a, nano::node_config & config_a, nano::logger_mt & logger_a, nano::online_reps & online_reps_a, nano::ledger & ledger_a, nano::network_params & network_params_a);
void vote (std::shared_ptr<nano::vote>, std::shared_ptr<nano::transport::channel>);
/** Note: node.active.mutex lock is required */
nano::vote_code vote_blocking (nano::transaction const &, std::shared_ptr<nano::vote>, std::shared_ptr<nano::transport::channel>, bool = false);
void verify_votes (std::deque<std::pair<std::shared_ptr<nano::vote>, std::shared_ptr<nano::transport::channel>>> &);
nano::vote_code vote_blocking (std::shared_ptr<nano::vote>, std::shared_ptr<nano::transport::channel>, bool = false);
void verify_votes (std::deque<std::pair<std::shared_ptr<nano::vote>, std::shared_ptr<nano::transport::channel>>> const &);
void flush ();
void calculate_weights ();
void stop ();
Expand All @@ -46,7 +46,6 @@ class vote_processor final

nano::signature_checker & checker;
nano::active_transactions & active;
nano::block_store & store;
nano::node_observers & observers;
nano::stat & stats;
nano::node_config & config;
Expand Down