From 1179b30aba96ee249b10c84a26782c88a9271f0f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Piotr=20Wo=CC=81jcik?=
 <3044353+pwojcikdev@users.noreply.github.com>
Date: Fri, 11 Oct 2024 15:46:30 +0200
Subject: [PATCH] Introduce bucketing component

---
 nano/core_test/CMakeLists.txt         |  1 +
 nano/core_test/bucketing.cpp          | 55 +++++++++++++++++++
 nano/core_test/election_scheduler.cpp |  3 +-
 nano/lib/numbers.hpp                  |  3 ++
 nano/node/CMakeLists.txt              |  2 +
 nano/node/bucketing.cpp               | 51 ++++++++++++++++++
 nano/node/bucketing.hpp               | 21 ++++++++
 nano/node/fwd.hpp                     |  1 +
 nano/node/node.cpp                    |  6 ++-
 nano/node/node.hpp                    |  2 +
 nano/node/scheduler/bucket.cpp        |  6 +--
 nano/node/scheduler/bucket.hpp        | 12 ++---
 nano/node/scheduler/component.cpp     |  4 +-
 nano/node/scheduler/component.hpp     |  2 +-
 nano/node/scheduler/priority.cpp      | 78 ++++++++-------------------
 nano/node/scheduler/priority.hpp      |  7 +--
 nano/secure/ledger.cpp                | 15 ++++++
 nano/secure/ledger.hpp                |  7 +++
 18 files changed, 201 insertions(+), 75 deletions(-)
 create mode 100644 nano/core_test/bucketing.cpp
 create mode 100644 nano/node/bucketing.cpp
 create mode 100644 nano/node/bucketing.hpp

diff --git a/nano/core_test/CMakeLists.txt b/nano/core_test/CMakeLists.txt
index bb4e6e9a0..dcc509212 100644
--- a/nano/core_test/CMakeLists.txt
+++ b/nano/core_test/CMakeLists.txt
@@ -12,6 +12,7 @@ add_executable(
   blockprocessor.cpp
   bootstrap.cpp
   bootstrap_server.cpp
+  bucketing.cpp
   cli.cpp
   confirmation_solicitor.cpp
   confirming_set.cpp
diff --git a/nano/core_test/bucketing.cpp b/nano/core_test/bucketing.cpp
new file mode 100644
index 000000000..94a6b4158
--- /dev/null
+++ b/nano/core_test/bucketing.cpp
@@ -0,0 +1,55 @@
+#include <nano/node/bucketing.hpp>
+
+#include <gtest/gtest.h>
+
+#include <algorithm>
+
+TEST (bucketing, construction)
+{
+	nano::bucketing bucketing;
+	ASSERT_EQ (63, bucketing.size ());
+}
+
+TEST (bucketing, zero_index)
+{
+	nano::bucketing bucketing;
+	ASSERT_EQ (0, bucketing.bucket_index (0));
+}
+
+TEST (bucketing, raw_index)
+{
+	nano::bucketing bucketing;
+	ASSERT_EQ (0, bucketing.bucket_index (nano::raw_ratio));
+}
+
+TEST (bucketing, nano_index)
+{
+	nano::bucketing bucketing;
+	ASSERT_EQ (14, bucketing.bucket_index (nano::nano_ratio));
+}
+
+TEST (bucketing, Knano_index)
+{
+	nano::bucketing bucketing;
+	ASSERT_EQ (49, bucketing.bucket_index (nano::Knano_ratio));
+}
+
+TEST (bucketing, max_index)
+{
+	nano::bucketing bucketing;
+	ASSERT_EQ (62, bucketing.bucket_index (std::numeric_limits<nano::amount::underlying_type>::max ()));
+}
+
+TEST (bucketing, indices)
+{
+	nano::bucketing bucketing;
+	auto indices = bucketing.bucket_indices ();
+	ASSERT_EQ (63, indices.size ());
+	ASSERT_EQ (indices.size (), bucketing.size ());
+
+	// Check that the indices are in ascending order
+	ASSERT_TRUE (std::adjacent_find (indices.begin (), indices.end (), [] (auto const & lhs, auto const & rhs) {
+		return lhs >= rhs;
+	})
+	== indices.end ());
+}
\ No newline at end of file
diff --git a/nano/core_test/election_scheduler.cpp b/nano/core_test/election_scheduler.cpp
index a17620b44..3e4567b70 100644
--- a/nano/core_test/election_scheduler.cpp
+++ b/nano/core_test/election_scheduler.cpp
@@ -256,8 +256,7 @@ TEST (election_scheduler_bucket, construction)
 	auto & node = *system.add_node ();
 
 	nano::scheduler::priority_bucket_config bucket_config;
-	nano::scheduler::bucket bucket{ nano::Knano_ratio, bucket_config, node.active, node.stats };
-	ASSERT_EQ (nano::Knano_ratio, bucket.minimum_balance);
+	nano::scheduler::bucket bucket{ 0, bucket_config, node.active, node.stats };
 	ASSERT_TRUE (bucket.empty ());
 	ASSERT_EQ (0, bucket.size ());
 }
diff --git a/nano/lib/numbers.hpp b/nano/lib/numbers.hpp
index 26b6f1444..822030587 100644
--- a/nano/lib/numbers.hpp
+++ b/nano/lib/numbers.hpp
@@ -24,6 +24,9 @@ nano::uint128_t const Knano_ratio = nano::uint128_t ("10000000000000000000000000
 nano::uint128_t const nano_ratio = nano::uint128_t ("1000000000000000000000000000000"); // 10^30 = 1 nano
 nano::uint128_t const raw_ratio = nano::uint128_t ("1"); // 10^0
 
+using bucket_index = uint64_t;
+using priority_timestamp = uint64_t; // Priority within the bucket
+
 class uint128_union
 {
 public:
diff --git a/nano/node/CMakeLists.txt b/nano/node/CMakeLists.txt
index f50649b04..86ef4bd84 100644
--- a/nano/node/CMakeLists.txt
+++ b/nano/node/CMakeLists.txt
@@ -22,6 +22,8 @@ add_library(
   bandwidth_limiter.cpp
   blockprocessor.hpp
   blockprocessor.cpp
+  bucketing.hpp
+  bucketing.cpp
   bootstrap_weights_beta.hpp
   bootstrap_weights_live.hpp
   bootstrap/account_sets.hpp
diff --git a/nano/node/bucketing.cpp b/nano/node/bucketing.cpp
new file mode 100644
index 000000000..271c1e1f8
--- /dev/null
+++ b/nano/node/bucketing.cpp
@@ -0,0 +1,51 @@
+#include <nano/lib/blocks.hpp>
+#include <nano/lib/utility.hpp>
+#include <nano/node/bucketing.hpp>
+#include <nano/secure/ledger.hpp>
+#include <nano/secure/ledger_set_any.hpp>
+
+nano::bucketing::bucketing ()
+{
+	auto build_region = [this] (uint128_t const & begin, uint128_t const & end, size_t count) {
+		auto width = (end - begin) / count;
+		for (auto i = 0; i < count; ++i)
+		{
+			minimums.push_back (begin + i * width);
+		}
+	};
+
+	minimums.push_back (uint128_t{ 0 });
+	build_region (uint128_t{ 1 } << 79, uint128_t{ 1 } << 88, 1);
+	build_region (uint128_t{ 1 } << 88, uint128_t{ 1 } << 92, 2);
+	build_region (uint128_t{ 1 } << 92, uint128_t{ 1 } << 96, 4);
+	build_region (uint128_t{ 1 } << 96, uint128_t{ 1 } << 100, 8);
+	build_region (uint128_t{ 1 } << 100, uint128_t{ 1 } << 104, 16);
+	build_region (uint128_t{ 1 } << 104, uint128_t{ 1 } << 108, 16);
+	build_region (uint128_t{ 1 } << 108, uint128_t{ 1 } << 112, 8);
+	build_region (uint128_t{ 1 } << 112, uint128_t{ 1 } << 116, 4);
+	build_region (uint128_t{ 1 } << 116, uint128_t{ 1 } << 120, 2);
+	minimums.push_back (uint128_t{ 1 } << 120);
+
+	for (auto i = 0; i < minimums.size (); ++i)
+	{
+		indices.push_back (i);
+	}
+}
+
+nano::bucket_index nano::bucketing::bucket_index (nano::amount balance) const
+{
+	release_assert (!minimums.empty ());
+	auto it = std::upper_bound (minimums.begin (), minimums.end (), balance);
+	release_assert (it != minimums.begin ()); // There should always be a bucket with a minimum_balance of 0
+	return std::distance (minimums.begin (), std::prev (it));
+}
+
+std::vector<nano::bucket_index> const & nano::bucketing::bucket_indices () const
+{
+	return indices;
+}
+
+size_t nano::bucketing::size () const
+{
+	return minimums.size ();
+}
\ No newline at end of file
diff --git a/nano/node/bucketing.hpp b/nano/node/bucketing.hpp
new file mode 100644
index 000000000..09a5bff03
--- /dev/null
+++ b/nano/node/bucketing.hpp
@@ -0,0 +1,21 @@
+#pragma once
+
+#include <nano/lib/numbers.hpp>
+#include <nano/node/fwd.hpp>
+
+namespace nano
+{
+class bucketing
+{
+public:
+	bucketing ();
+
+	nano::bucket_index bucket_index (nano::amount balance) const;
+	std::vector<nano::bucket_index> const & bucket_indices () const;
+	size_t size () const;
+
+private:
+	std::vector<nano::uint128_t> minimums;
+	std::vector<nano::bucket_index> indices;
+};
+}
\ No newline at end of file
diff --git a/nano/node/fwd.hpp b/nano/node/fwd.hpp
index b7b3740e1..414c3e6c3 100644
--- a/nano/node/fwd.hpp
+++ b/nano/node/fwd.hpp
@@ -10,6 +10,7 @@ namespace nano
 class account_sets_config;
 class active_elections;
 class block_processor;
+class bucketing;
 class bootstrap_config;
 class bootstrap_server;
 class bootstrap_service;
diff --git a/nano/node/node.cpp b/nano/node/node.cpp
index 16d5a6f8c..9dea5c770 100644
--- a/nano/node/node.cpp
+++ b/nano/node/node.cpp
@@ -14,6 +14,7 @@
 #include <nano/node/bootstrap/bootstrap_service.hpp>
 #include <nano/node/bootstrap_weights_beta.hpp>
 #include <nano/node/bootstrap_weights_live.hpp>
+#include <nano/node/bucketing.hpp>
 #include <nano/node/confirming_set.hpp>
 #include <nano/node/daemonconfig.hpp>
 #include <nano/node/election_status.hpp>
@@ -129,6 +130,8 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
 	block_processor{ *block_processor_impl },
 	confirming_set_impl{ std::make_unique<nano::confirming_set> (config.confirming_set, ledger, block_processor, stats, logger) },
 	confirming_set{ *confirming_set_impl },
+	bucketing_impl{ std::make_unique<nano::bucketing> () },
+	bucketing{ *bucketing_impl },
 	active_impl{ std::make_unique<nano::active_elections> (*this, confirming_set, block_processor) },
 	active{ *active_impl },
 	rep_crawler (config.rep_crawler, *this),
@@ -149,7 +152,7 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
 	generator{ *generator_impl },
 	final_generator_impl{ std::make_unique<nano::vote_generator> (config, *this, ledger, wallets, vote_processor, history, network, stats, logger, /* final */ true) },
 	final_generator{ *final_generator_impl },
-	scheduler_impl{ std::make_unique<nano::scheduler::component> (config, *this, ledger, block_processor, active, online_reps, vote_cache, confirming_set, stats, logger) },
+	scheduler_impl{ std::make_unique<nano::scheduler::component> (config, *this, ledger, bucketing, block_processor, active, online_reps, vote_cache, confirming_set, stats, logger) },
 	scheduler{ *scheduler_impl },
 	aggregator_impl{ std::make_unique<nano::request_aggregator> (config.request_aggregator, *this, stats, generator, final_generator, history, ledger, wallets, vote_router) },
 	aggregator{ *aggregator_impl },
@@ -317,6 +320,7 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
 		logger.info (nano::log::type::node, "Work pool threads: {} ({})", work.threads.size (), (work.opencl ? "OpenCL" : "CPU"));
 		logger.info (nano::log::type::node, "Work peers: {}", config.work_peers.size ());
 		logger.info (nano::log::type::node, "Node ID: {}", node_id.pub.to_node_id ());
+		logger.info (nano::log::type::node, "Number of buckets: {}", bucketing.size ());
 
 		if (!work_generation_enabled ())
 		{
diff --git a/nano/node/node.hpp b/nano/node/node.hpp
index 2d4f4a351..8b6830bc7 100644
--- a/nano/node/node.hpp
+++ b/nano/node/node.hpp
@@ -179,6 +179,8 @@ public:
 	nano::block_processor & block_processor;
 	std::unique_ptr<nano::confirming_set> confirming_set_impl;
 	nano::confirming_set & confirming_set;
+	std::unique_ptr<nano::bucketing> bucketing_impl;
+	nano::bucketing & bucketing;
 	std::unique_ptr<nano::active_elections> active_impl;
 	nano::active_elections & active;
 	nano::online_reps online_reps;
diff --git a/nano/node/scheduler/bucket.cpp b/nano/node/scheduler/bucket.cpp
index 31a0152b3..4e4ecb0a5 100644
--- a/nano/node/scheduler/bucket.cpp
+++ b/nano/node/scheduler/bucket.cpp
@@ -8,9 +8,9 @@
  * bucket
  */
 
-nano::scheduler::bucket::bucket (nano::uint128_t minimum_balance_a, priority_bucket_config const & config_a, nano::active_elections & active_a, nano::stats & stats_a) :
+nano::scheduler::bucket::bucket (nano::bucket_index index_a, priority_bucket_config const & config_a, nano::active_elections & active_a, nano::stats & stats_a) :
+	index{ index_a },
 	config{ config_a },
-	minimum_balance{ minimum_balance_a },
 	active{ active_a },
 	stats{ stats_a }
 {
@@ -34,7 +34,7 @@ bool nano::scheduler::bucket::available () const
 	}
 }
 
-bool nano::scheduler::bucket::election_vacancy (priority_t candidate) const
+bool nano::scheduler::bucket::election_vacancy (nano::priority_timestamp candidate) const
 {
 	debug_assert (!mutex.try_lock ());
 
diff --git a/nano/node/scheduler/bucket.hpp b/nano/node/scheduler/bucket.hpp
index e9b6d48d8..b5b12b2b8 100644
--- a/nano/node/scheduler/bucket.hpp
+++ b/nano/node/scheduler/bucket.hpp
@@ -47,14 +47,12 @@ public:
 class bucket final
 {
 public:
-	using priority_t = uint64_t;
+	nano::bucket_index const index;
 
 public:
-	bucket (nano::uint128_t minimum_balance, priority_bucket_config const &, nano::active_elections &, nano::stats &);
+	bucket (nano::bucket_index, priority_bucket_config const &, nano::active_elections &, nano::stats &);
 	~bucket ();
 
-	nano::uint128_t const minimum_balance;
-
 	bool available () const;
 	bool activate ();
 	void update ();
@@ -70,7 +68,7 @@ public:
 	void dump () const;
 
 private:
-	bool election_vacancy (priority_t candidate) const;
+	bool election_vacancy (nano::priority_timestamp candidate) const;
 	bool election_overfill () const;
 	void cancel_lowest_election ();
 
@@ -118,7 +116,7 @@ private: // Elections
 	{
 		std::shared_ptr<nano::election> election;
 		nano::qualified_root root;
-		priority_t priority;
+		nano::priority_timestamp priority;
 	};
 
 	// clang-format off
@@ -128,7 +126,7 @@ private: // Elections
 		mi::hashed_unique<mi::tag<tag_root>,
 			mi::member<election_entry, nano::qualified_root, &election_entry::root>>,
 		mi::ordered_non_unique<mi::tag<tag_priority>,
-			mi::member<election_entry, priority_t, &election_entry::priority>>
+			mi::member<election_entry, nano::priority_timestamp, &election_entry::priority>>
 	>>;
 	// clang-format on
 
diff --git a/nano/node/scheduler/component.cpp b/nano/node/scheduler/component.cpp
index 2de155605..82b034bd1 100644
--- a/nano/node/scheduler/component.cpp
+++ b/nano/node/scheduler/component.cpp
@@ -5,11 +5,11 @@
 #include <nano/node/scheduler/optimistic.hpp>
 #include <nano/node/scheduler/priority.hpp>
 
-nano::scheduler::component::component (nano::node_config & node_config, nano::node & node, nano::ledger & ledger, nano::block_processor & block_processor, nano::active_elections & active, nano::online_reps & online_reps, nano::vote_cache & vote_cache, nano::confirming_set & confirming_set, nano::stats & stats, nano::logger & logger) :
+nano::scheduler::component::component (nano::node_config & node_config, nano::node & node, nano::ledger & ledger, nano::bucketing & bucketing, nano::block_processor & block_processor, nano::active_elections & active, nano::online_reps & online_reps, nano::vote_cache & vote_cache, nano::confirming_set & confirming_set, nano::stats & stats, nano::logger & logger) :
 	hinted_impl{ std::make_unique<nano::scheduler::hinted> (node_config.hinted_scheduler, node, vote_cache, active, online_reps, stats) },
 	manual_impl{ std::make_unique<nano::scheduler::manual> (node) },
 	optimistic_impl{ std::make_unique<nano::scheduler::optimistic> (node_config.optimistic_scheduler, node, ledger, active, node_config.network_params.network, stats) },
-	priority_impl{ std::make_unique<nano::scheduler::priority> (node_config, node, ledger, block_processor, active, confirming_set, stats, logger) },
+	priority_impl{ std::make_unique<nano::scheduler::priority> (node_config, node, ledger, bucketing, block_processor, active, confirming_set, stats, logger) },
 	hinted{ *hinted_impl },
 	manual{ *manual_impl },
 	optimistic{ *optimistic_impl },
diff --git a/nano/node/scheduler/component.hpp b/nano/node/scheduler/component.hpp
index 1af64ff00..97373577d 100644
--- a/nano/node/scheduler/component.hpp
+++ b/nano/node/scheduler/component.hpp
@@ -10,7 +10,7 @@ namespace nano::scheduler
 class component final
 {
 public:
-	component (nano::node_config &, nano::node &, nano::ledger &, nano::block_processor &, nano::active_elections &, nano::online_reps &, nano::vote_cache &, nano::confirming_set &, nano::stats &, nano::logger &);
+	component (nano::node_config &, nano::node &, nano::ledger &, nano::bucketing &, nano::block_processor &, nano::active_elections &, nano::online_reps &, nano::vote_cache &, nano::confirming_set &, nano::stats &, nano::logger &);
 	~component ();
 
 	void start ();
diff --git a/nano/node/scheduler/priority.cpp b/nano/node/scheduler/priority.cpp
index a40816d2d..a05d9a44e 100644
--- a/nano/node/scheduler/priority.cpp
+++ b/nano/node/scheduler/priority.cpp
@@ -1,5 +1,6 @@
 #include <nano/lib/blocks.hpp>
 #include <nano/node/active_elections.hpp>
+#include <nano/node/bucketing.hpp>
 #include <nano/node/election.hpp>
 #include <nano/node/node.hpp>
 #include <nano/node/scheduler/priority.hpp>
@@ -7,44 +8,20 @@
 #include <nano/secure/ledger_set_any.hpp>
 #include <nano/secure/ledger_set_confirmed.hpp>
 
-nano::scheduler::priority::priority (nano::node_config & node_config, nano::node & node_a, nano::ledger & ledger_a, nano::block_processor & block_processor_a, nano::active_elections & active_a, nano::confirming_set & confirming_set_a, nano::stats & stats_a, nano::logger & logger_a) :
+nano::scheduler::priority::priority (nano::node_config & node_config, nano::node & node_a, nano::ledger & ledger_a, nano::bucketing & bucketing_a, nano::block_processor & block_processor_a, nano::active_elections & active_a, nano::confirming_set & confirming_set_a, nano::stats & stats_a, nano::logger & logger_a) :
 	config{ node_config.priority_scheduler },
 	node{ node_a },
 	ledger{ ledger_a },
+	bucketing{ bucketing_a },
 	block_processor{ block_processor_a },
 	active{ active_a },
 	confirming_set{ confirming_set_a },
 	stats{ stats_a },
 	logger{ logger_a }
 {
-	std::vector<nano::uint128_t> minimums;
-
-	auto build_region = [&minimums] (uint128_t const & begin, uint128_t const & end, size_t count) {
-		auto width = (end - begin) / count;
-		for (auto i = 0; i < count; ++i)
-		{
-			minimums.push_back (begin + i * width);
-		}
-	};
-
-	minimums.push_back (uint128_t{ 0 });
-	build_region (uint128_t{ 1 } << 79, uint128_t{ 1 } << 88, 1);
-	build_region (uint128_t{ 1 } << 88, uint128_t{ 1 } << 92, 2);
-	build_region (uint128_t{ 1 } << 92, uint128_t{ 1 } << 96, 4);
-	build_region (uint128_t{ 1 } << 96, uint128_t{ 1 } << 100, 8);
-	build_region (uint128_t{ 1 } << 100, uint128_t{ 1 } << 104, 16);
-	build_region (uint128_t{ 1 } << 104, uint128_t{ 1 } << 108, 16);
-	build_region (uint128_t{ 1 } << 108, uint128_t{ 1 } << 112, 8);
-	build_region (uint128_t{ 1 } << 112, uint128_t{ 1 } << 116, 4);
-	build_region (uint128_t{ 1 } << 116, uint128_t{ 1 } << 120, 2);
-	minimums.push_back (uint128_t{ 1 } << 120);
-
-	logger.debug (nano::log::type::election_scheduler, "Number of buckets: {}", minimums.size ());
-
-	for (size_t i = 0u, n = minimums.size (); i < n; ++i)
+	for (auto const & index : bucketing.bucket_indices ())
 	{
-		auto bucket = std::make_unique<scheduler::bucket> (minimums[i], node_config.priority_bucket, active, stats);
-		buckets.emplace_back (std::move (bucket));
+		buckets[index] = std::make_unique<scheduler::bucket> (index, node_config.priority_bucket, active, stats);
 	}
 
 	// Activate accounts with fresh blocks
@@ -141,14 +118,14 @@ bool nano::scheduler::priority::activate (secure::transaction const & transactio
 
 	if (ledger.dependents_confirmed (transaction, *block))
 	{
-		auto const balance = block->balance ();
-		auto const previous_balance = ledger.any.block_balance (transaction, conf_info.frontier).value_or (0);
-		auto const balance_priority = std::max (balance, previous_balance);
+		auto const [priority_balance, priority_timestamp] = ledger.block_priority (transaction, *block);
+		auto const bucket_index = bucketing.bucket_index (priority_balance);
 
 		bool added = false;
 		{
-			auto & bucket = find_bucket (balance_priority);
-			added = bucket.push (account_info.modified, block);
+			auto const & bucket = buckets.at (bucket_index);
+			release_assert (bucket);
+			added = bucket->push (account_info.modified, block);
 		}
 		if (added)
 		{
@@ -157,7 +134,8 @@ bool nano::scheduler::priority::activate (secure::transaction const & transactio
 			nano::log::arg{ "account", account.to_account () }, // TODO: Convert to lazy eval
 			nano::log::arg{ "block", block },
 			nano::log::arg{ "time", account_info.modified },
-			nano::log::arg{ "priority", balance_priority });
+			nano::log::arg{ "priority_balance", priority_balance },
+			nano::log::arg{ "priority_timestamp", priority_timestamp });
 
 			notify ();
 		}
@@ -187,7 +165,7 @@ bool nano::scheduler::priority::activate_successors (secure::transaction const &
 bool nano::scheduler::priority::contains (nano::block_hash const & hash) const
 {
 	return std::any_of (buckets.begin (), buckets.end (), [&hash] (auto const & bucket) {
-		return bucket->contains (hash);
+		return bucket.second->contains (hash);
 	});
 }
 
@@ -199,21 +177,21 @@ void nano::scheduler::priority::notify ()
 std::size_t nano::scheduler::priority::size () const
 {
 	return std::accumulate (buckets.begin (), buckets.end (), std::size_t{ 0 }, [] (auto const & sum, auto const & bucket) {
-		return sum + bucket->size ();
+		return sum + bucket.second->size ();
 	});
 }
 
 bool nano::scheduler::priority::empty () const
 {
 	return std::all_of (buckets.begin (), buckets.end (), [] (auto const & bucket) {
-		return bucket->empty ();
+		return bucket.second->empty ();
 	});
 }
 
 bool nano::scheduler::priority::predicate () const
 {
 	return std::any_of (buckets.begin (), buckets.end (), [] (auto const & bucket) {
-		return bucket->available ();
+		return bucket.second->available ();
 	});
 }
 
@@ -232,7 +210,7 @@ void nano::scheduler::priority::run ()
 
 			lock.unlock ();
 
-			for (auto & bucket : buckets)
+			for (auto const & [index, bucket] : buckets)
 			{
 				if (bucket->available ())
 				{
@@ -259,7 +237,7 @@ void nano::scheduler::priority::run_cleanup ()
 
 			lock.unlock ();
 
-			for (auto & bucket : buckets)
+			for (auto const & [index, bucket] : buckets)
 			{
 				bucket->update ();
 			}
@@ -269,34 +247,22 @@ void nano::scheduler::priority::run_cleanup ()
 	}
 }
 
-auto nano::scheduler::priority::find_bucket (nano::uint128_t priority) -> bucket &
-{
-	auto it = std::upper_bound (buckets.begin (), buckets.end (), priority, [] (nano::uint128_t const & priority, std::unique_ptr<bucket> const & bucket) {
-		return priority < bucket->minimum_balance;
-	});
-	release_assert (it != buckets.begin ()); // There should always be a bucket with a minimum_balance of 0
-	it = std::prev (it);
-	return **it;
-}
-
 nano::container_info nano::scheduler::priority::container_info () const
 {
 	auto collect_blocks = [&] () {
 		nano::container_info info;
-		for (auto i = 0; i < buckets.size (); ++i)
+		for (auto const & [index, bucket] : buckets)
 		{
-			auto const & bucket = buckets[i];
-			info.put (std::to_string (i), bucket->size ());
+			info.put (std::to_string (index), bucket->size ());
 		}
 		return info;
 	};
 
 	auto collect_elections = [&] () {
 		nano::container_info info;
-		for (auto i = 0; i < buckets.size (); ++i)
+		for (auto const & [index, bucket] : buckets)
 		{
-			auto const & bucket = buckets[i];
-			info.put (std::to_string (i), bucket->election_count ());
+			info.put (std::to_string (index), bucket->election_count ());
 		}
 		return info;
 	};
diff --git a/nano/node/scheduler/priority.hpp b/nano/node/scheduler/priority.hpp
index 808528164..6f0e3badd 100644
--- a/nano/node/scheduler/priority.hpp
+++ b/nano/node/scheduler/priority.hpp
@@ -6,6 +6,7 @@
 
 #include <condition_variable>
 #include <deque>
+#include <map>
 #include <memory>
 #include <string>
 #include <thread>
@@ -26,7 +27,7 @@ public:
 class priority final
 {
 public:
-	priority (nano::node_config &, nano::node &, nano::ledger &, nano::block_processor &, nano::active_elections &, nano::confirming_set &, nano::stats &, nano::logger &);
+	priority (nano::node_config &, nano::node &, nano::ledger &, nano::bucketing &, nano::block_processor &, nano::active_elections &, nano::confirming_set &, nano::stats &, nano::logger &);
 	~priority ();
 
 	void start ();
@@ -51,6 +52,7 @@ private: // Dependencies
 	priority_config const & config;
 	nano::node & node;
 	nano::ledger & ledger;
+	nano::bucketing & bucketing;
 	nano::block_processor & block_processor;
 	nano::active_elections & active;
 	nano::confirming_set & confirming_set;
@@ -61,10 +63,9 @@ private:
 	void run ();
 	void run_cleanup ();
 	bool predicate () const;
-	bucket & find_bucket (nano::uint128_t priority);
 
 private:
-	std::vector<std::unique_ptr<bucket>> buckets;
+	std::map<nano::bucket_index, std::unique_ptr<scheduler::bucket>> buckets;
 
 	bool stopped{ false };
 	nano::condition_variable condition;
diff --git a/nano/secure/ledger.cpp b/nano/secure/ledger.cpp
index 23550e743..b46b188f2 100644
--- a/nano/secure/ledger.cpp
+++ b/nano/secure/ledger.cpp
@@ -1254,6 +1254,21 @@ uint64_t nano::ledger::pruning_action (secure::write_transaction & transaction_a
 	return pruned_count;
 }
 
+auto nano::ledger::block_priority (nano::secure::transaction const & transaction, nano::block const & block) const -> block_priority_result
+{
+	auto const balance = block.balance ();
+	auto const previous_block = !block.previous ().is_zero () ? any.block_get (transaction, block.previous ()) : nullptr;
+	auto const previous_balance = previous_block ? previous_block->balance () : 0;
+
+	// Handle full send case nicely where the balance would otherwise be 0
+	auto const priority_balance = std::max (balance, block.is_send () ? previous_balance : 0);
+
+	// Use previous block timestamp as priority timestamp for least recently used prioritization within the same bucket
+	// Account info timestamp is not used here because it will get out of sync when rollbacks happen
+	auto const priority_timestamp = previous_block ? previous_block->sideband ().timestamp : block.sideband ().timestamp;
+	return { priority_balance, priority_timestamp };
+}
+
 // A precondition is that the store is an LMDB store
 bool nano::ledger::migrate_lmdb_to_rocksdb (std::filesystem::path const & data_path_a) const
 {
diff --git a/nano/secure/ledger.hpp b/nano/secure/ledger.hpp
index 69c863780..61b79aa22 100644
--- a/nano/secure/ledger.hpp
+++ b/nano/secure/ledger.hpp
@@ -75,13 +75,20 @@ public:
 	nano::link const & epoch_link (nano::epoch) const;
 	bool migrate_lmdb_to_rocksdb (std::filesystem::path const &) const;
 	bool bootstrap_weight_reached () const;
+
 	static nano::epoch version (nano::block const & block);
 	nano::epoch version (secure::transaction const &, nano::block_hash const & hash) const;
+
 	uint64_t cemented_count () const;
 	uint64_t block_count () const;
 	uint64_t account_count () const;
 	uint64_t pruned_count () 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
+	using block_priority_result = std::pair<nano::amount, nano::priority_timestamp>;
+	block_priority_result block_priority (secure::transaction const &, nano::block const &) const;
+
 	nano::container_info container_info () const;
 
 public: