Unlimited backlog when bootstrapping (#4922)

nano/node/block_processor.cpp

@@ -212,12 +212,13 @@ void nano::block_processor::rollback_competitor (secure::write_transaction & tra
 
 double nano::block_processor::backlog_factor () const
 {
-	auto const backlog = ledger.backlog_count ();
-	if (node_config.max_backlog == 0 || backlog <= node_config.max_backlog * config.backlog_threshold)
+	auto const backlog = ledger.backlog_size ();
+	auto const max_backlog = ledger.max_backlog ();
+	if (max_backlog == 0 || backlog <= max_backlog * config.backlog_threshold)
 	{
 		return 0.0;
 	}
-	return std::max (1.0, static_cast<double> (backlog) / static_cast<double> (node_config.max_backlog * config.backlog_threshold));
+	return std::max (1.0, static_cast<double> (backlog) / static_cast<double> (max_backlog * config.backlog_threshold));
 }
 
 void nano::block_processor::wait_backlog (nano::unique_lock<nano::mutex> & lock)

nano/node/bounded_backlog.cpp

@@ -26,7 +26,7 @@ nano::bounded_backlog::bounded_backlog (nano::node_config const & config_a, nano
 	logger{ logger_a },
 	scan_limiter{ config.bounded_backlog.scan_rate }
 {
-	if (!config.bounded_backlog.enable || config.max_backlog == 0)
+	if (!config.bounded_backlog.enable || ledger.max_backlog () == 0)
 	{
 		return;
 	}
@@ -95,7 +95,7 @@ void nano::bounded_backlog::start ()
 {
 	debug_assert (!thread.joinable ());
 
-	if (!config.bounded_backlog.enable || config.max_backlog == 0)
+	if (!config.bounded_backlog.enable || ledger.max_backlog () == 0)
 	{
 		return;
 	}
@@ -193,10 +193,12 @@ bool nano::bounded_backlog::insert (nano::secure::transaction const & transactio
 bool nano::bounded_backlog::predicate () const
 {
 	debug_assert (!mutex.try_lock ());
-	debug_assert (config.max_backlog > 0); // Should be fully disabled if max_backlog is 0
 
 	// Both ledger and tracked backlog must be over the threshold
-	return ledger.backlog_count () > config.max_backlog && index.size () > config.max_backlog;
+	auto const max_backlog = ledger.max_backlog ();
+	debug_assert (max_backlog > 0); // Should be fully disabled if max_backlog is 0
+
+	return ledger.backlog_size () > max_backlog && index.size () > max_backlog;
 }
 
 void nano::bounded_backlog::run ()
@@ -225,10 +227,17 @@ void nano::bounded_backlog::run ()
 		stats.inc (nano::stat::type::bounded_backlog, nano::stat::detail::loop);
 
 		// Calculate the number of targets to rollback
-		uint64_t const backlog = ledger.backlog_count ();
-		uint64_t const target_count = backlog > config.max_backlog ? backlog - config.max_backlog : 0;
+		auto const backlog = ledger.backlog_size ();
+		auto const max_backlog = ledger.max_backlog ();
+		uint64_t const target_count = backlog > max_backlog ? backlog - max_backlog : 0;
 
-		auto targets = gather_targets (std::min (target_count, static_cast<uint64_t> (config.bounded_backlog.batch_size)));
+		if (target_count == 0)
+		{
+			continue;
+		}
+
+		auto const bucket_threshold = max_backlog / bucketing.size ();
+		auto targets = gather_targets (std::min (target_count, static_cast<uint64_t> (config.bounded_backlog.batch_size)), bucket_threshold);
 		if (!targets.empty ())
 		{
 			lock.unlock ();
@@ -350,12 +359,7 @@ std::deque<nano::block_hash> nano::bounded_backlog::perform_rollbacks (std::dequ
 	return processed;
 }
 
-size_t nano::bounded_backlog::bucket_threshold () const
-{
-	return config.max_backlog / bucketing.size ();
-}
-
-std::deque<nano::block_hash> nano::bounded_backlog::gather_targets (size_t max_count) const
+std::deque<nano::block_hash> nano::bounded_backlog::gather_targets (size_t max_count, size_t bucket_threshold) const
 {
 	debug_assert (!mutex.try_lock ());
 
@@ -365,7 +369,7 @@ std::deque<nano::block_hash> nano::bounded_backlog::gather_targets (size_t max_c
 	for (auto bucket : bucketing.bucket_indices ())
 	{
 		// Only start rolling back if the bucket is over the threshold of unconfirmed blocks
-		if (index.size (bucket) > bucket_threshold ())
+		if (index.size (bucket) > bucket_threshold)
 		{
 			auto const count = std::min (max_count, config.bounded_backlog.batch_size);
 

nano/node/bounded_backlog.hpp

@@ -115,7 +115,6 @@ public:
 	void stop ();
 
 	size_t index_size () const;
-	size_t bucket_threshold () const;
 	bool contains (nano::block_hash const &) const;
 
 	nano::container_info container_info () const;
@@ -138,7 +137,7 @@ private:
 
 	bool predicate () const;
 	void run ();
-	std::deque<nano::block_hash> gather_targets (size_t max_count) const;
+	std::deque<nano::block_hash> gather_targets (size_t max_count, size_t bucket_threshold) const;
 	bool should_rollback (nano::block_hash const &) const;
 
 	std::deque<nano::block_hash> perform_rollbacks (std::deque<nano::block_hash> const & targets, size_t max_rollbacks);

nano/node/node.cpp

@@ -102,7 +102,7 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
 	wallets_store{ *wallets_store_impl },
 	wallets_impl{ std::make_unique<nano::wallets> (wallets_store.init_error (), *this) },
 	wallets{ *wallets_impl },
-	ledger_impl{ std::make_unique<nano::ledger> (store, network_params.ledger, stats, logger, flags_a.generate_cache, config_a.representative_vote_weight_minimum.number ()) },
+	ledger_impl{ std::make_unique<nano::ledger> (store, network_params.ledger, stats, logger, flags_a.generate_cache, config.representative_vote_weight_minimum.number (), config.max_backlog) },
 	ledger{ *ledger_impl },
 	runner_impl{ std::make_unique<nano::thread_runner> (io_ctx_shared, logger, config.io_threads) },
 	runner{ *runner_impl },

nano/secure/ledger.cpp

@@ -30,12 +30,13 @@
 
 #include <cryptopp/words.h>
 
-nano::ledger::ledger (nano::store::component & store_a, nano::ledger_constants & constants_a, nano::stats & stats_a, nano::logger & logger_a, nano::generate_cache_flags generate_cache_flags_a, nano::uint128_t min_rep_weight_a) :
+nano::ledger::ledger (nano::store::component & store_a, nano::ledger_constants & constants_a, nano::stats & stats_a, nano::logger & logger_a, nano::generate_cache_flags generate_cache_flags_a, nano::uint128_t min_rep_weight_a, uint64_t max_backlog_a) :
 	store{ store_a },
 	constants{ constants_a },
 	stats{ stats_a },
 	logger{ logger_a },
 	rep_weights{ store_a.rep_weight, min_rep_weight_a },
+	max_backlog_size{ max_backlog_a },
 	any_impl{ std::make_unique<ledger_set_any> (*this) },
 	confirmed_impl{ std::make_unique<ledger_set_confirmed> (*this) },
 	any{ *any_impl },
@@ -1069,13 +1070,37 @@ uint64_t nano::ledger::pruned_count () const
 	return cache.pruned_count;
 }
 
-uint64_t nano::ledger::backlog_count () const
+uint64_t nano::ledger::backlog_size () const
 {
 	auto blocks = cache.block_count.load ();
 	auto cemented = cache.cemented_count.load ();
 	return (blocks > cemented) ? blocks - cemented : 0;
 }
 
+uint64_t nano::ledger::max_backlog () const
+{
+	auto const count = cemented_count ();
+	auto const max_bootstrap_count = bootstrap_weight_max_blocks;
+
+	if (max_backlog_size == 0)
+	{
+		return 0; // Unlimited backlog
+	}
+
+	// Use cemented block count to determine the switch point for backlog
+	if (count >= max_bootstrap_count)
+	{
+		return max_backlog_size;
+	}
+	else
+	{
+		// If the bootstrap weight hasn't been reached, we allow a backlog of up to bootstrap_weight_max_blocks
+		// This should avoid having to rollback too many blocks once the bootstrap weight is reached
+		auto const allowed_backlog = max_bootstrap_count - count;
+		return std::max (allowed_backlog, max_backlog_size);
+	}
+}
+
 nano::container_info nano::ledger::container_info () const
 {
 	nano::container_info info;

nano/secure/ledger.hpp

@@ -37,7 +37,7 @@ class ledger final
 	friend class receivable_iterator;
 
 public:
-	ledger (nano::store::component &, nano::ledger_constants &, nano::stats &, nano::logger &, nano::generate_cache_flags = {}, nano::uint128_t min_rep_weight = 0);
+	ledger (nano::store::component &, nano::ledger_constants &, nano::stats &, nano::logger &, nano::generate_cache_flags = {}, nano::uint128_t min_rep_weight = 0, uint64_t max_backlog = 0);
 	~ledger ();
 
 	/** Start read-write transaction */
@@ -86,7 +86,8 @@ public:
 	uint64_t block_count () const;
 	uint64_t account_count () const;
 	uint64_t pruned_count () const;
-	uint64_t backlog_count () const;
+	uint64_t backlog_size () const;
+	uint64_t max_backlog () const;
 
 	// Returned priority balance is maximum of block balance and previous block balance to handle full account balance send cases
 	// Returned timestamp is the previous block timestamp or the current timestamp if there's no previous block
@@ -109,6 +110,8 @@ public:
 	nano::rep_weights rep_weights;
 
 public:
+	uint64_t const max_backlog_size{ 0 };
+
 	std::unordered_map<nano::account, nano::uint128_t> bootstrap_weights;
 	uint64_t bootstrap_weight_max_blocks{ 1 };