Broadcast wire txs and blocks on demand.

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.
     /// -----------------------------------------------------------------------
 

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_{};
 };

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.

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<network::wire_tx>(message, {});
+}
+
+void chaser::broadcast(const network::wire_block::cptr& message) const NOEXCEPT
+{
+    node_.broadcast<network::wire_block>(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.
 // ----------------------------------------------------------------------------
 

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<const chain::view::block>(
+        archive().get_wire_block(link, true), true);
+
+    if (block->is_valid())
+        broadcast(emplace_shared<const wire_block>(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

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<const chain::view::transaction>(
+        tx.to_data(true), false, true);
+    broadcast(emplace_shared<const network::wire_tx>(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

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()

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<network::wire_block::cptr> promise{};
+    auto broadcast = promise.get_future();
+    const auto session = std::make_shared<wire_session>(node_, 42);
+    session->subscribe<network::wire_block>([&](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(); }));

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<network::wire_tx::cptr> promise{};
+    auto broadcast = promise.get_future();
+    const auto session = std::make_shared<wire_session>(node_, 42);
+    session->subscribe<network::wire_tx>([&](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);