Move block observer logic into active_transactions (#4421)

nano/node/CMakeLists.txt

@@ -24,8 +24,6 @@ add_library(
   block_arrival.cpp
   block_broadcast.cpp
   block_broadcast.hpp
-  block_publisher.cpp
-  block_publisher.hpp
   blocking_observer.cpp
   blocking_observer.hpp
   blockprocessor.hpp

nano/node/active_transactions.cpp

@@ -13,9 +13,10 @@
 
 using namespace std::chrono;
 
-nano::active_transactions::active_transactions (nano::node & node_a, nano::confirmation_height_processor & confirmation_height_processor_a) :
-	confirmation_height_processor{ confirmation_height_processor_a },
+nano::active_transactions::active_transactions (nano::node & node_a, nano::confirmation_height_processor & confirmation_height_processor_a, nano::block_processor & block_processor_a) :
 	node{ node_a },
+	confirmation_height_processor{ confirmation_height_processor_a },
+	block_processor{ block_processor_a },
 	recently_confirmed{ 65536 },
 	recently_cemented{ node.config.confirmation_history_size },
 	election_time_to_live{ node_a.network_params.network.is_dev_network () ? 0s : 2s }
@@ -31,6 +32,18 @@ nano::active_transactions::active_transactions (nano::node & node_a, nano::confi
 	confirmation_height_processor.add_block_already_cemented_observer ([this] (nano::block_hash const & hash_a) {
 		this->block_already_cemented_callback (hash_a);
 	});
+
+	// Notify elections about alternative (forked) blocks
+	block_processor.processed.add ([this] (auto const & result, auto const & block) {
+		switch (result.code)
+		{
+			case nano::process_result::fork:
+				publish (block);
+				break;
+			default:
+				break;
+		}
+	});
 }
 
 nano::active_transactions::~active_transactions ()

nano/node/active_transactions.hpp

@@ -25,6 +25,7 @@ class node;
 class active_transactions;
 class block;
 class block_sideband;
+class block_processor;
 class election;
 class vote;
 class confirmation_height_processor;
@@ -130,7 +131,7 @@ private: // Elections
 	std::unordered_map<nano::block_hash, std::shared_ptr<nano::election>> blocks;
 
 public:
-	active_transactions (nano::node &, nano::confirmation_height_processor &);
+	active_transactions (nano::node &, nano::confirmation_height_processor &, nano::block_processor &);
 	~active_transactions ();
 
 	void start ();
@@ -204,8 +205,9 @@ private:
 	void notify_observers (nano::election_status const & status, std::vector<nano::vote_with_weight_info> const & votes, nano::account const & account, nano::uint128_t amount, bool is_state_send, bool is_state_epoch, nano::account const & pending_account);
 
 private: // Dependencies
-	nano::confirmation_height_processor & confirmation_height_processor;
 	nano::node & node;
+	nano::confirmation_height_processor & confirmation_height_processor;
+	nano::block_processor & block_processor;
 
 public:
 	recently_confirmed_cache recently_confirmed;

nano/node/block_publisher.cpp

@@ -1,27 +0,0 @@
-#include <nano/node/active_transactions.hpp>
-#include <nano/node/block_publisher.hpp>
-#include <nano/node/blockprocessor.hpp>
-
-nano::block_publisher::block_publisher (nano::active_transactions & active) :
-	active{ active }
-{
-}
-
-void nano::block_publisher::connect (nano::block_processor & block_processor)
-{
-	block_processor.processed.add ([this] (auto const & result, auto const & block) {
-		switch (result.code)
-		{
-			case nano::process_result::fork:
-				observe (block);
-				break;
-			default:
-				break;
-		}
-	});
-}
-
-void nano::block_publisher::observe (std::shared_ptr<nano::block> block)
-{
-	active.publish (block);
-}

nano/node/block_publisher.hpp

@@ -1,24 +0,0 @@
-#pragma once
-
-#include <memory>
-
-namespace nano
-{
-class active_transactions;
-class block_processor;
-class block;
-
-// This class tracks processed blocks to be published.
-class block_publisher
-{
-public:
-	block_publisher (nano::active_transactions & active);
-	void connect (nano::block_processor & block_processor);
-
-private:
-	// Block_processor observer
-	void observe (std::shared_ptr<nano::block> block);
-
-	nano::active_transactions & active;
-};
-}

nano/node/node.cpp

@@ -186,7 +186,7 @@ nano::node::node (boost::asio::io_context & io_ctx_a, std::filesystem::path cons
 	vote_cache{ config.vote_cache, stats },
 	generator{ config, ledger, wallets, vote_processor, history, network, stats, logger, /* non-final */ false },
 	final_generator{ config, ledger, wallets, vote_processor, history, network, stats, logger, /* final */ true },
-	active (*this, confirmation_height_processor),
+	active{ *this, confirmation_height_processor, block_processor },
 	scheduler_impl{ std::make_unique<nano::scheduler::component> (*this) },
 	scheduler{ *scheduler_impl },
 	aggregator (config, stats, generator, final_generator, history, ledger, wallets, active),
@@ -198,14 +198,13 @@ nano::node::node (boost::asio::io_context & io_ctx_a, std::filesystem::path cons
 	startup_time (std::chrono::steady_clock::now ()),
 	node_seq (seq),
 	block_broadcast{ network, block_arrival, !flags.disable_block_processor_republishing },
-	block_publisher{ active },
 	process_live_dispatcher{ ledger, scheduler.priority, vote_cache, websocket }
 {
 	logger.debug (nano::log::type::node, "Constructing node...");
 
 	block_broadcast.connect (block_processor);
-	block_publisher.connect (block_processor);
 	process_live_dispatcher.connect (block_processor);
+
 	unchecked.satisfied.add ([this] (nano::unchecked_info const & info) {
 		this->block_processor.add (info.block);
 	});

nano/node/node.hpp

@@ -10,7 +10,6 @@
 #include <nano/node/bandwidth_limiter.hpp>
 #include <nano/node/block_arrival.hpp>
 #include <nano/node/block_broadcast.hpp>
-#include <nano/node/block_publisher.hpp>
 #include <nano/node/blockprocessor.hpp>
 #include <nano/node/bootstrap/bootstrap.hpp>
 #include <nano/node/bootstrap/bootstrap_attempt.hpp>
@@ -191,7 +190,6 @@ public:
 	nano::websocket_server websocket;
 	nano::epoch_upgrader epoch_upgrader;
 	nano::block_broadcast block_broadcast;
-	nano::block_publisher block_publisher;
 	nano::process_live_dispatcher process_live_dispatcher;
 
 	std::chrono::steady_clock::time_point const startup_time;