diff --git a/include/bitcoin/node/chasers/chaser.hpp b/include/bitcoin/node/chasers/chaser.hpp index c2b6bc64..bc6169fc 100644 --- a/include/bitcoin/node/chasers/chaser.hpp +++ b/include/bitcoin/node/chasers/chaser.hpp @@ -107,6 +107,19 @@ class BCN_API chaser virtual void notify_one(object_key key, const code& ec, event_value value) const NOEXCEPT; + /// Broadcasts. + /// ----------------------------------------------------------------------- + + /// Broadcast a wire tx|block to subscribed channels. + virtual void broadcast( + const network::wire_tx::cptr& message) const NOEXCEPT; + virtual void broadcast( + const network::wire_block::cptr& message) const NOEXCEPT; + + /// Some channel demands wire tx|block broadcasts. + virtual bool wire_tx_demanded() const NOEXCEPT; + virtual bool wire_block_demanded() const NOEXCEPT; + /// Strand. /// ----------------------------------------------------------------------- diff --git a/include/bitcoin/node/chasers/chaser_confirm.hpp b/include/bitcoin/node/chasers/chaser_confirm.hpp index 8c83396a..48021e52 100644 --- a/include/bitcoin/node/chasers/chaser_confirm.hpp +++ b/include/bitcoin/node/chasers/chaser_confirm.hpp @@ -40,6 +40,7 @@ class BCN_API chaser_confirm void stopping(const code& ec) NOEXCEPT override; protected: + using chaser::broadcast; using header_link = database::header_link; using header_links = database::header_links; using header_states = database::header_states; @@ -70,10 +71,15 @@ class BCN_API chaser_confirm bool roll_back(const header_links& popped, size_t fork_point, size_t top) NOEXCEPT; void announce(const header_link& link, height_t height) NOEXCEPT; + void broadcast(const header_link& link, bool organized) NOEXCEPT; + void do_broadcast(const header_link& link, bool organized) NOEXCEPT; void do_stopping(const code& ec) NOEXCEPT; void start_stale_timer() NOEXCEPT; void handle_stale_timer(const code& ec) NOEXCEPT; + // This is thread safe. + network::asio::strand broadcast_strand_; + // This is protected by strand. network::deadline::ptr stale_timer_{}; }; diff --git a/include/bitcoin/node/chasers/chaser_transaction.hpp b/include/bitcoin/node/chasers/chaser_transaction.hpp index 4c92a186..78869100 100644 --- a/include/bitcoin/node/chasers/chaser_transaction.hpp +++ b/include/bitcoin/node/chasers/chaser_transaction.hpp @@ -52,6 +52,8 @@ class BCN_API chaser_transaction submit_handler&& handler) NOEXCEPT; protected: + using chaser::broadcast; + virtual bool handle_chase(const code& ec, event_value value) NOEXCEPT; virtual void do_submit(const system::chain::transactions_cptr& txs, @@ -90,6 +92,7 @@ class BCN_API chaser_transaction const system::chain::context& ctx) NOEXCEPT; code unconfirmable(const database::header_link& link, size_t height, const code& ec) NOEXCEPT; + void broadcast(const system::chain::transaction& tx) const NOEXCEPT; void start_scan(bool remap) NOEXCEPT; // These are thread safe. diff --git a/src/chasers/chaser.cpp b/src/chasers/chaser.cpp index 3e0ea12c..6928098d 100644 --- a/src/chasers/chaser.cpp +++ b/src/chasers/chaser.cpp @@ -104,6 +104,29 @@ void chaser::notify_one(object_key key, const code& ec, node_.notify_one(key, ec, value); } +// Broadcasts. +// ---------------------------------------------------------------------------- + +void chaser::broadcast(const network::wire_tx::cptr& message) const NOEXCEPT +{ + node_.broadcast(message, {}); +} + +void chaser::broadcast(const network::wire_block::cptr& message) const NOEXCEPT +{ + node_.broadcast(message, {}); +} + +bool chaser::wire_tx_demanded() const NOEXCEPT +{ + return node_.wire_tx_demanded(); +} + +bool chaser::wire_block_demanded() const NOEXCEPT +{ + return node_.wire_block_demanded(); +} + // Strand. // ---------------------------------------------------------------------------- diff --git a/src/chasers/chaser_confirm.cpp b/src/chasers/chaser_confirm.cpp index 966d8517..b578602b 100644 --- a/src/chasers/chaser_confirm.cpp +++ b/src/chasers/chaser_confirm.cpp @@ -36,7 +36,8 @@ using namespace std::placeholders; BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) chaser_confirm::chaser_confirm(full_node& node) NOEXCEPT - : chaser(node) + : chaser(node), + broadcast_strand_(node.service().get_executor()) { } @@ -378,6 +379,7 @@ bool chaser_confirm::set_reorganized(const header_link& link, return false; notify(error::success, chases::reorganized{ link }); + broadcast(link, false); fire(events::block_reorganized, confirmed_height); LOGV("Block reorganized: " << confirmed_height); return true; @@ -411,6 +413,7 @@ bool chaser_confirm::set_organized(const header_link& link, return false; notify(error::success, chases::organized{ link }); + broadcast(link, true); fire(events::block_organized, confirmed_height); LOGV("Block organized: " << confirmed_height); @@ -447,6 +450,30 @@ void chaser_confirm::announce(const header_link& link, height_t) NOEXCEPT } } +void chaser_confirm::broadcast(const header_link& link, + bool organized) NOEXCEPT +{ + BC_ASSERT(stranded()); + + if (wire_block_demanded() && is_current_chain(true)) + boost::asio::post(broadcast_strand_, + BIND(do_broadcast, link, organized)); +} + +void chaser_confirm::do_broadcast(const header_link& link, + bool organized) NOEXCEPT +{ + BC_ASSERT(broadcast_strand_.running_in_this_thread()); + if (closed()) + return; + + const auto block = emplace_shared( + archive().get_wire_block(link, true), true); + + if (block->is_valid()) + broadcast(emplace_shared(block, organized)); +} + // The chain becomes stale at the top block timestamp plus the window, which // may be sooner than the window, as the block may have been mined earlier. void chaser_confirm::start_stale_timer() NOEXCEPT diff --git a/src/chasers/chaser_transaction.cpp b/src/chasers/chaser_transaction.cpp index bf2675ce..ff7eb0b1 100644 --- a/src/chasers/chaser_transaction.cpp +++ b/src/chasers/chaser_transaction.cpp @@ -228,6 +228,7 @@ void chaser_transaction::do_reorganized(header_t link) NOEXCEPT } notify(error::success, chases::transaction{ fk }); + broadcast(*tx); } } @@ -327,6 +328,7 @@ void chaser_transaction::do_submit(const transactions_cptr& txs, bool test, fire(events::tx_archived, to_rate(tx)); notify(error::success, chases::transaction{ link }); + broadcast(tx); } handler(error::success, {}); @@ -454,6 +456,7 @@ code chaser_transaction::fill(database::tx_link& out, return fault(error::transaction6); notify(error::success, chases::transaction{ out }); + broadcast(tx); return error::success; } @@ -481,11 +484,22 @@ code chaser_transaction::fill(database::tx_link& out, { fire(events::tx_archived, to_rate(tx)); notify(error::success, chases::transaction{ out }); + broadcast(tx); } return error::success; } +void chaser_transaction::broadcast(const chain::transaction& tx) const NOEXCEPT +{ + if (!wire_tx_demanded()) + return; + + const auto view = emplace_shared( + tx.to_data(true), false, true); + broadcast(emplace_shared(view)); +} + // An identified block with an invalid tx is invalid. code chaser_transaction::unconfirmable(const database::header_link& link, size_t height, const code& ec) NOEXCEPT diff --git a/test/chasers/chaser.cpp b/test/chasers/chaser.cpp index ae4b1ad7..3fab5587 100644 --- a/test/chasers/chaser.cpp +++ b/test/chasers/chaser.cpp @@ -45,6 +45,8 @@ class chaser_accessor using chaser::is_current_header; using chaser::is_recent; using chaser::suspended; + using chaser::wire_tx_demanded; + using chaser::wire_block_demanded; }; // A running full node without peer connections, and a chaser over it. @@ -165,6 +167,30 @@ BOOST_AUTO_TEST_CASE(chaser__is_recent__genesis__false) BOOST_REQUIRE(!chaser_.is_recent()); } +BOOST_AUTO_TEST_CASE(chaser__wire_tx_demanded__demanded__true) +{ + BOOST_REQUIRE(!chaser_.wire_tx_demanded()); + + node_.demand_wire_tx(true); + BOOST_REQUIRE(chaser_.wire_tx_demanded()); + BOOST_REQUIRE(!chaser_.wire_block_demanded()); + + node_.demand_wire_tx(false); + BOOST_REQUIRE(!chaser_.wire_tx_demanded()); +} + +BOOST_AUTO_TEST_CASE(chaser__wire_block_demanded__demanded__true) +{ + BOOST_REQUIRE(!chaser_.wire_block_demanded()); + + node_.demand_wire_block(true); + BOOST_REQUIRE(chaser_.wire_block_demanded()); + BOOST_REQUIRE(!chaser_.wire_tx_demanded()); + + node_.demand_wire_block(false); + BOOST_REQUIRE(!chaser_.wire_block_demanded()); +} + BOOST_AUTO_TEST_SUITE_END() BC_POP_WARNING() diff --git a/test/chasers/chaser_confirm.cpp b/test/chasers/chaser_confirm.cpp index 5c7445a4..668ad24e 100644 --- a/test/chasers/chaser_confirm.cpp +++ b/test/chasers/chaser_confirm.cpp @@ -223,11 +223,30 @@ static bool confirm_c1_confirmed(node::query& query) return confirm_confirmed(query, confirm_c1(), 1); } +static bool confirm_genesis_only(node::query&) +{ + return true; +} + static void confirm_uncheckpointed(configuration& config) { config.bitcoin.checkpoints.clear(); } +static void confirm_current(configuration& config) +{ + config.bitcoin.checkpoints.clear(); + config.node.currency_window_minutes = 30u * 365u * 24u * 60u; +} + +// A session to observe wire broadcasts. +class wire_session + : public network::session_manual +{ +public: + using session_manual::session_manual; +}; + struct confirm_setup_fixture { DELETE_COPY_MOVE(confirm_setup_fixture); @@ -457,6 +476,15 @@ struct confirm_unprevouted_fixture } }; +struct confirm_genesis_fixture + : confirm_setup_fixture +{ + confirm_genesis_fixture() + : confirm_setup_fixture(confirm_genesis_only, confirm_current) + { + } +}; + BOOST_AUTO_TEST_SUITE(chaser_confirm_tests) ////BOOST_FIXTURE_TEST_CASE(chaser_confirm__start__stronger_candidate_fork__reorganized, confirm_stronger_fixture) @@ -538,6 +566,31 @@ BOOST_FIXTURE_TEST_CASE(chaser_confirm__start__bypassed_candidate_unheaded_paren BOOST_REQUIRE_EQUAL(query_.get_top_confirmed(), 1u); } +BOOST_FIXTURE_TEST_CASE(chaser_confirm__bump__wire_block_demanded_current__broadcast_block, confirm_genesis_fixture) +{ + std::promise promise{}; + auto broadcast = promise.get_future(); + const auto session = std::make_shared(node_, 42); + session->subscribe([&](const code&, const network::wire_block::cptr& message, uint64_t) NOEXCEPT + { + promise.set_value(message); + return false; + }, 42); + + node_.demand_wire_block(true); + BOOST_REQUIRE(confirm_confirmable(query_)); + + node_.notify(node::error::success, chases::bump{ 0 }); + BOOST_REQUIRE(confirmed(confirm_d1(), 1)); + BOOST_REQUIRE(broadcast.wait_for(std::chrono::seconds(10)) == std::future_status::ready); + + const auto message = broadcast.get(); + BOOST_REQUIRE(message->organized()); + BOOST_REQUIRE(message->block()->to_data(true) == confirm_d1().to_data(true)); + BOOST_REQUIRE_EQUAL(message->transactions().size(), 1u); + BOOST_REQUIRE(message->transactions().front()->tx()->to_data(true) == confirm_d1().transactions_ptr()->front()->to_data(true)); +} + ////BOOST_FIXTURE_TEST_CASE(chaser_confirm__start__valid_candidate_without_prevouts__suspended_unconfirmable, confirm_unprevouted_fixture) ////{ //// BOOST_REQUIRE(await([&]() { return node_.suspended(); })); diff --git a/test/chasers/chaser_transaction.cpp b/test/chasers/chaser_transaction.cpp index ac99c8a2..a1be69c8 100644 --- a/test/chasers/chaser_transaction.cpp +++ b/test/chasers/chaser_transaction.cpp @@ -122,6 +122,14 @@ static bool stored_header(node::query& query) NOEXCEPT BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT) +// A session to observe wire broadcasts. +class wire_session + : public network::session_manual +{ +public: + using session_manual::session_manual; +}; + struct chaser_transaction_setup_fixture : p2p_setup_fixture { @@ -462,6 +470,24 @@ BOOST_FIXTURE_TEST_CASE(chaser_transaction__submit__valid__success_archived, cha BOOST_REQUIRE(query_.is_tx(tx->hash(false))); } +BOOST_FIXTURE_TEST_CASE(chaser_transaction__submit__wire_tx_demanded__broadcast, chaser_transaction_pooling_setup_fixture) +{ + std::promise promise{}; + auto broadcast = promise.get_future(); + const auto session = std::make_shared(node_, 42); + session->subscribe([&](const code&, const network::wire_tx::cptr& message, uint64_t) NOEXCEPT + { + promise.set_value(message); + return false; + }, 42); + + const auto tx = spend(parent_value); + node_.demand_wire_tx(true); + BOOST_REQUIRE_EQUAL(submit(package({ tx })).first, node::error::success); + BOOST_REQUIRE(broadcast.wait_for(std::chrono::seconds(10)) == std::future_status::ready); + BOOST_REQUIRE(broadcast.get()->tx()->to_data(true) == tx->to_data(true)); +} + BOOST_FIXTURE_TEST_CASE(chaser_transaction__submit__valid_twice__duplicate_transaction, chaser_transaction_pooling_setup_fixture) { const auto tx = spend(parent_value);