diff --git a/nano/core_test/active_elections.cpp b/nano/core_test/active_elections.cpp
index d36d764a5b..a52659c6e7 100644
--- a/nano/core_test/active_elections.cpp
+++ b/nano/core_test/active_elections.cpp
@@ -669,7 +669,7 @@ TEST (active_elections, dropped_cleanup)
 	ASSERT_FALSE (node.network.publish_filter.apply (block_bytes.data (), block_bytes.size ()));
 
 	// An election was recently dropped
-	ASSERT_EQ (1, node.stats.count (nano::stat::type::active_dropped, nano::stat::detail::normal));
+	ASSERT_EQ (1, node.stats.count (nano::stat::type::active_dropped, nano::stat::detail::manual));
 
 	// Block cleared from active
 	ASSERT_FALSE (node.vote_router.active (hash));
@@ -687,7 +687,7 @@ TEST (active_elections, dropped_cleanup)
 	ASSERT_TRUE (node.network.publish_filter.apply (block_bytes.data (), block_bytes.size ()));
 
 	// Not dropped
-	ASSERT_EQ (1, node.stats.count (nano::stat::type::active_dropped, nano::stat::detail::normal));
+	ASSERT_EQ (1, node.stats.count (nano::stat::type::active_dropped, nano::stat::detail::manual));
 
 	// Block cleared from active
 	ASSERT_FALSE (node.vote_router.active (hash));
@@ -1305,125 +1305,6 @@ TEST (active_elections, list_active)
 	auto active = node.active.list_active ();
 }
 
-TEST (active_elections, vacancy)
-{
-	std::atomic<bool> updated = false;
-	{
-		nano::test::system system;
-		nano::node_config config = system.default_config ();
-		config.active_elections.size = 1;
-		auto & node = *system.add_node (config);
-		nano::state_block_builder builder;
-		auto send = builder.make_block ()
-					.account (nano::dev::genesis_key.pub)
-					.previous (nano::dev::genesis->hash ())
-					.representative (nano::dev::genesis_key.pub)
-					.link (nano::dev::genesis_key.pub)
-					.balance (nano::dev::constants.genesis_amount - nano::Gxrb_ratio)
-					.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
-					.work (*system.work.generate (nano::dev::genesis->hash ()))
-					.build ();
-		node.active.vacancy_update = [&updated] () { updated = true; };
-		ASSERT_EQ (nano::block_status::progress, node.process (send));
-		ASSERT_EQ (1, node.active.vacancy ());
-		ASSERT_EQ (0, node.active.size ());
-		node.scheduler.priority.activate (node.ledger.tx_begin_read (), nano::dev::genesis_key.pub);
-		ASSERT_TIMELY (1s, updated);
-		updated = false;
-		ASSERT_EQ (0, node.active.vacancy ());
-		ASSERT_EQ (1, node.active.size ());
-		auto election1 = node.active.election (send->qualified_root ());
-		ASSERT_NE (nullptr, election1);
-		election1->force_confirm ();
-		ASSERT_TIMELY (1s, updated);
-		ASSERT_EQ (1, node.active.vacancy ());
-		ASSERT_EQ (0, node.active.size ());
-	}
-}
-
-// Ensure transactions in excess of capacity are removed in fifo order
-TEST (active_elections, fifo)
-{
-	nano::test::system system{};
-
-	nano::node_config config = system.default_config ();
-	config.active_elections.size = 1;
-	config.frontiers_confirmation = nano::frontiers_confirmation_mode::disabled;
-
-	auto & node = *system.add_node (config);
-	auto latest_hash = nano::dev::genesis->hash ();
-	nano::keypair key0{};
-	nano::state_block_builder builder{};
-
-	// Construct two pending entries that can be received simultaneously
-	auto send1 = builder.make_block ()
-				 .previous (latest_hash)
-				 .account (nano::dev::genesis_key.pub)
-				 .representative (nano::dev::genesis_key.pub)
-				 .link (key0.pub)
-				 .balance (nano::dev::constants.genesis_amount - 1)
-				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
-				 .work (*system.work.generate (latest_hash))
-				 .build ();
-	ASSERT_EQ (nano::block_status::progress, node.process (send1));
-	node.process_confirmed (nano::election_status{ send1 });
-	ASSERT_TIMELY (5s, node.block_confirmed (send1->hash ()));
-
-	nano::keypair key1{};
-	latest_hash = send1->hash ();
-	auto send2 = builder.make_block ()
-				 .previous (latest_hash)
-				 .account (nano::dev::genesis_key.pub)
-				 .representative (nano::dev::genesis_key.pub)
-				 .link (key1.pub)
-				 .balance (nano::dev::constants.genesis_amount - 2)
-				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
-				 .work (*system.work.generate (latest_hash))
-				 .build ();
-	ASSERT_EQ (nano::block_status::progress, node.process (send2));
-	node.process_confirmed (nano::election_status{ send2 });
-	ASSERT_TIMELY (5s, node.block_confirmed (send2->hash ()));
-
-	auto receive1 = builder.make_block ()
-					.previous (0)
-					.account (key0.pub)
-					.representative (nano::dev::genesis_key.pub)
-					.link (send1->hash ())
-					.balance (1)
-					.sign (key0.prv, key0.pub)
-					.work (*system.work.generate (key0.pub))
-					.build ();
-	ASSERT_EQ (nano::block_status::progress, node.process (receive1));
-
-	auto receive2 = builder.make_block ()
-					.previous (0)
-					.account (key1.pub)
-					.representative (nano::dev::genesis_key.pub)
-					.link (send2->hash ())
-					.balance (1)
-					.sign (key1.prv, key1.pub)
-					.work (*system.work.generate (key1.pub))
-					.build ();
-	ASSERT_EQ (nano::block_status::progress, node.process (receive2));
-
-	// Ensure first transaction becomes active
-	node.scheduler.manual.push (receive1);
-	ASSERT_TIMELY (5s, node.active.election (receive1->qualified_root ()) != nullptr);
-
-	// Ensure second transaction becomes active
-	node.scheduler.manual.push (receive2);
-	ASSERT_TIMELY (5s, node.active.election (receive2->qualified_root ()) != nullptr);
-
-	// Ensure excess transactions get trimmed
-	ASSERT_TIMELY_EQ (5s, node.active.size (), 1);
-
-	// Ensure overflow stats have been incremented
-	ASSERT_EQ (1, node.stats.count (nano::stat::type::active_dropped, nano::stat::detail::normal));
-
-	// Ensure the surviving transaction is the least recently inserted
-	ASSERT_TIMELY (1s, node.active.election (receive2->qualified_root ()) != nullptr);
-}
-
 /*
  * Ensures we limit the number of vote hinted elections in AEC
  */
@@ -1487,103 +1368,6 @@ TEST (active_elections, limit_vote_hinted_elections)
 	ASSERT_TIMELY (5s, nano::test::active (node, { open1 }));
 
 	// Ensure there was no overflow of elections
-	ASSERT_EQ (0, node.stats.count (nano::stat::type::active_dropped, nano::stat::detail::normal));
+	ASSERT_EQ (0, node.stats.count (nano::stat::type::active_dropped, nano::stat::detail::priority));
 }
 
-/*
- * Tests that when AEC is running at capacity from normal elections, it is still possible to schedule a limited number of hinted elections
- */
-TEST (active_elections, allow_limited_overflow)
-{
-	nano::test::system system;
-	nano::node_config config = system.default_config ();
-	const int aec_limit = 20;
-	config.frontiers_confirmation = nano::frontiers_confirmation_mode::disabled;
-	config.active_elections.size = aec_limit;
-	config.active_elections.hinted_limit_percentage = 20; // Should give us a limit of 4 hinted elections
-	auto & node = *system.add_node (config);
-
-	auto blocks = nano::test::setup_independent_blocks (system, node, aec_limit * 4);
-
-	// Split blocks in two halves
-	std::vector<std::shared_ptr<nano::block>> blocks1 (blocks.begin (), blocks.begin () + blocks.size () / 2);
-	std::vector<std::shared_ptr<nano::block>> blocks2 (blocks.begin () + blocks.size () / 2, blocks.end ());
-
-	// Even though automatic frontier confirmation is disabled, AEC is doing funny stuff and inserting elections, clear that
-	WAIT (1s);
-	node.active.clear ();
-	ASSERT_TRUE (node.active.empty ());
-
-	// Insert the first part of the blocks into normal election scheduler
-	for (auto const & block : blocks1)
-	{
-		node.scheduler.priority.activate (node.ledger.tx_begin_read (), block->account ());
-	}
-
-	// Ensure number of active elections reaches AEC limit and there is no overfill
-	ASSERT_TIMELY_EQ (5s, node.active.size (), node.active.limit ());
-	// And it stays that way without increasing
-	ASSERT_ALWAYS (1s, node.active.size () == node.active.limit ());
-
-	// Insert votes for the second part of the blocks, so that those are scheduled as hinted elections
-	for (auto const & block : blocks2)
-	{
-		// Non-final vote, so it stays in the AEC without getting confirmed
-		auto vote = nano::test::make_vote (nano::dev::genesis_key, { block });
-		node.vote_cache.insert (vote);
-	}
-
-	// Ensure active elections overfill AEC only up to normal + hinted limit
-	ASSERT_TIMELY_EQ (5s, node.active.size (), node.active.limit () + node.active.limit (nano::election_behavior::hinted));
-	// And it stays that way without increasing
-	ASSERT_ALWAYS (1s, node.active.size () == node.active.limit () + node.active.limit (nano::election_behavior::hinted));
-}
-
-/*
- * Tests that when hinted elections are present in the AEC, normal scheduler adapts not to exceed the limit of all elections
- */
-TEST (active_elections, allow_limited_overflow_adapt)
-{
-	nano::test::system system;
-	nano::node_config config = system.default_config ();
-	const int aec_limit = 20;
-	config.frontiers_confirmation = nano::frontiers_confirmation_mode::disabled;
-	config.active_elections.size = aec_limit;
-	config.active_elections.hinted_limit_percentage = 20; // Should give us a limit of 4 hinted elections
-	auto & node = *system.add_node (config);
-
-	auto blocks = nano::test::setup_independent_blocks (system, node, aec_limit * 4);
-
-	// Split blocks in two halves
-	std::vector<std::shared_ptr<nano::block>> blocks1 (blocks.begin (), blocks.begin () + blocks.size () / 2);
-	std::vector<std::shared_ptr<nano::block>> blocks2 (blocks.begin () + blocks.size () / 2, blocks.end ());
-
-	// Even though automatic frontier confirmation is disabled, AEC is doing funny stuff and inserting elections, clear that
-	WAIT (1s);
-	node.active.clear ();
-	ASSERT_TRUE (node.active.empty ());
-
-	// Insert votes for the second part of the blocks, so that those are scheduled as hinted elections
-	for (auto const & block : blocks2)
-	{
-		// Non-final vote, so it stays in the AEC without getting confirmed
-		auto vote = nano::test::make_vote (nano::dev::genesis_key, { block });
-		node.vote_cache.insert (vote);
-	}
-
-	// Ensure hinted election amount is bounded by hinted limit
-	ASSERT_TIMELY_EQ (5s, node.active.size (), node.active.limit (nano::election_behavior::hinted));
-	// And it stays that way without increasing
-	ASSERT_ALWAYS (1s, node.active.size () == node.active.limit (nano::election_behavior::hinted));
-
-	// Insert the first part of the blocks into normal election scheduler
-	for (auto const & block : blocks1)
-	{
-		node.scheduler.priority.activate (node.ledger.tx_begin_read (), block->account ());
-	}
-
-	// Ensure number of active elections reaches AEC limit and there is no overfill
-	ASSERT_TIMELY_EQ (5s, node.active.size (), node.active.limit ());
-	// And it stays that way without increasing
-	ASSERT_ALWAYS (1s, node.active.size () == node.active.limit ());
-}
diff --git a/nano/core_test/blockprocessor.cpp b/nano/core_test/blockprocessor.cpp
index 57c777062b..4b4a4d98bc 100644
--- a/nano/core_test/blockprocessor.cpp
+++ b/nano/core_test/blockprocessor.cpp
@@ -1,9 +1,12 @@
 #include <nano/lib/blockbuilders.hpp>
 #include <nano/lib/blocks.hpp>
+#include <nano/node/active_elections.hpp>
+#include <nano/node/election.hpp>
 #include <nano/node/node.hpp>
 #include <nano/node/nodeconfig.hpp>
 #include <nano/secure/common.hpp>
 #include <nano/secure/ledger.hpp>
+#include <nano/test_common/chains.hpp>
 #include <nano/test_common/system.hpp>
 #include <nano/test_common/testutil.hpp>
 
@@ -40,3 +43,117 @@ TEST (block_processor, broadcast_block_on_arrival)
 	// Checks whether the block was broadcast.
 	ASSERT_TIMELY (5s, node2->block_or_pruned_exists (send1->hash ()));
 }
+
+TEST (block_processor, rollback_overflow)
+{
+	nano::test::system system;
+	nano::node_config config;
+	config.priority_scheduler.depth = 1;
+	auto node = system.add_node (config);
+	nano::state_block_builder builder;
+	nano::keypair key1;
+	auto send1 = builder.make_block ()
+				 .account (nano::dev::genesis_key.pub)
+				 .previous (nano::dev::genesis->hash ())
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::dev::constants.genesis_amount - nano::Gxrb_ratio)
+				 .link (key1.pub)
+				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+				 .work (*system.work.generate (nano::dev::genesis->hash ()))
+				 .build ();
+	ASSERT_EQ (nano::block_status::progress, node->ledger.process (node->ledger.tx_begin_write (), send1));
+	node->ledger.confirm (node->ledger.tx_begin_write (), send1->hash ());
+	nano::keypair key2;
+	auto send2 = builder.make_block ()
+				 .account (nano::dev::genesis_key.pub)
+				 .previous (send1->hash ())
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::dev::constants.genesis_amount - 2 * nano::Gxrb_ratio)
+				 .link (key2.pub)
+				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+				 .work (*system.work.generate (send1->hash ()))
+				 .build ();
+	ASSERT_EQ (nano::block_status::progress, node->ledger.process (node->ledger.tx_begin_write (), send2));
+	node->ledger.confirm (node->ledger.tx_begin_write (), send2->hash ());
+
+	auto open1 = builder.make_block ()
+				 .account (key1.pub)
+				 .previous (0)
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::Gxrb_ratio)
+				 .link (send1->hash ())
+				 .sign (key1.prv, key1.pub)
+				 .work (*system.work.generate (key1.pub))
+				 .build ();
+	std::optional<nano::block_status> status;
+	ASSERT_TRUE ((status = node->block_processor.add_blocking (open1, nano::block_source::live), status && status.value () == nano::block_status::progress));
+	auto open2 = builder.make_block ()
+				 .account (key2.pub)
+				 .previous (0)
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::Gxrb_ratio)
+				 .link (send2->hash ())
+				 .sign (key2.prv, key2.pub)
+				 .work (*system.work.generate (key2.pub))
+				 .build ();
+	ASSERT_TRUE ((status = node->block_processor.add_blocking (open2, nano::block_source::live), status && status.value () == nano::block_status::overflow));
+}
+
+TEST (block_processor, scheduler_confirmed_space)
+{
+	nano::test::system system;
+	nano::node_config config;
+	config.priority_scheduler.depth = 1;
+	auto node = system.add_node (config);
+	nano::state_block_builder builder;
+	nano::keypair key1;
+	auto send1 = builder.make_block ()
+				 .account (nano::dev::genesis_key.pub)
+				 .previous (nano::dev::genesis->hash ())
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::dev::constants.genesis_amount - nano::Gxrb_ratio)
+				 .link (key1.pub)
+				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+				 .work (*system.work.generate (nano::dev::genesis->hash ()))
+				 .build ();
+	ASSERT_EQ (nano::block_status::progress, node->ledger.process (node->ledger.tx_begin_write (), send1));
+	node->ledger.confirm (node->ledger.tx_begin_write (), send1->hash ());
+	nano::keypair key2;
+	auto send2 = builder.make_block ()
+				 .account (nano::dev::genesis_key.pub)
+				 .previous (send1->hash ())
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::dev::constants.genesis_amount - 2 * nano::Gxrb_ratio)
+				 .link (key2.pub)
+				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+				 .work (*system.work.generate (send1->hash ()))
+				 .build ();
+	ASSERT_EQ (nano::block_status::progress, node->ledger.process (node->ledger.tx_begin_write (), send2));
+	node->ledger.confirm (node->ledger.tx_begin_write (), send2->hash ());
+
+	auto open1 = builder.make_block ()
+				 .account (key1.pub)
+				 .previous (0)
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::Gxrb_ratio)
+				 .link (send1->hash ())
+				 .sign (key1.prv, key1.pub)
+				 .work (*system.work.generate (key1.pub))
+				 .build ();
+	std::optional<nano::block_status> status;
+	ASSERT_TRUE ((status = node->block_processor.add_blocking (open1, nano::block_source::live), status && status.value () == nano::block_status::progress));
+	auto election = node->active.election (open1->qualified_root ());
+	ASSERT_NE (nullptr, election);
+	election->force_confirm ();
+	ASSERT_TIMELY (5s, node->active.empty ());
+	auto open2 = builder.make_block ()
+				 .account (key2.pub)
+				 .previous (0)
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::Gxrb_ratio)
+				 .link (send2->hash ())
+				 .sign (key2.prv, key2.pub)
+				 .work (*system.work.generate (key2.pub))
+				 .build ();
+	ASSERT_TRUE ((status = node->block_processor.add_blocking (open2, nano::block_source::live), status && status.value () == nano::block_status::progress));
+}
diff --git a/nano/core_test/confirmation_solicitor.cpp b/nano/core_test/confirmation_solicitor.cpp
index f3609a36a6..ca48a2ee8a 100644
--- a/nano/core_test/confirmation_solicitor.cpp
+++ b/nano/core_test/confirmation_solicitor.cpp
@@ -46,11 +46,11 @@ TEST (confirmation_solicitor, batches)
 		nano::lock_guard<nano::mutex> guard (node2.active.mutex);
 		for (size_t i (0); i < nano::network::confirm_req_hashes_max; ++i)
 		{
-			auto election (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::normal));
+			auto election (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::priority));
 			ASSERT_FALSE (solicitor.add (*election));
 		}
 		// Reached the maximum amount of requests for the channel
-		auto election (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::normal));
+		auto election (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::priority));
 		// Broadcasting should be immediate
 		ASSERT_EQ (0, node2.stats.count (nano::stat::type::message, nano::stat::detail::publish, nano::stat::dir::out));
 		ASSERT_FALSE (solicitor.broadcast (*election));
@@ -92,7 +92,7 @@ TEST (confirmation_solicitor, different_hash)
 				.work (*system.work.generate (nano::dev::genesis->hash ()))
 				.build ();
 	send->sideband_set ({});
-	auto election (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::normal));
+	auto election (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::priority));
 	// Add a vote for something else, not the winner
 	election->last_votes[representative.account] = { std::chrono::steady_clock::now (), 1, 1 };
 	// Ensure the request and broadcast goes through
@@ -136,7 +136,7 @@ TEST (confirmation_solicitor, bypass_max_requests_cap)
 				.work (*system.work.generate (nano::dev::genesis->hash ()))
 				.build ();
 	send->sideband_set ({});
-	auto election (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::normal));
+	auto election (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::priority));
 	// Add a vote for something else, not the winner
 	for (auto const & rep : representatives)
 	{
@@ -149,7 +149,7 @@ TEST (confirmation_solicitor, bypass_max_requests_cap)
 	ASSERT_TIMELY_EQ (6s, max_representatives + 1, node2.stats.count (nano::stat::type::message, nano::stat::detail::confirm_req, nano::stat::dir::out));
 
 	solicitor.prepare (representatives);
-	auto election2 (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::normal));
+	auto election2 (std::make_shared<nano::election> (node2, send, nullptr, nullptr, nano::election_behavior::priority));
 	ASSERT_FALSE (solicitor.add (*election2));
 	ASSERT_FALSE (solicitor.broadcast (*election2));
 
diff --git a/nano/core_test/election.cpp b/nano/core_test/election.cpp
index c285e12c6d..82bcb7f131 100644
--- a/nano/core_test/election.cpp
+++ b/nano/core_test/election.cpp
@@ -18,7 +18,7 @@ TEST (election, construction)
 	nano::test::system system (1);
 	auto & node = *system.nodes[0];
 	auto election = std::make_shared<nano::election> (
-	node, nano::dev::genesis, [] (auto const &) {}, [] (auto const &) {}, nano::election_behavior::normal);
+	node, nano::dev::genesis, [] (auto const &) {}, [] (auto const &) {}, nano::election_behavior::priority);
 }
 
 TEST (election, behavior)
@@ -27,7 +27,7 @@ TEST (election, behavior)
 	auto chain = nano::test::setup_chain (system, *system.nodes[0], 1, nano::dev::genesis_key, false);
 	auto election = nano::test::start_election (system, *system.nodes[0], chain[0]->hash ());
 	ASSERT_NE (nullptr, election);
-	ASSERT_EQ (nano::election_behavior::normal, election->behavior ());
+	ASSERT_EQ (nano::election_behavior::manual, election->behavior ());
 }
 
 TEST (election, quorum_minimum_flip_success)
diff --git a/nano/core_test/election_scheduler.cpp b/nano/core_test/election_scheduler.cpp
index 02a37bec89..8c525e9e1a 100644
--- a/nano/core_test/election_scheduler.cpp
+++ b/nano/core_test/election_scheduler.cpp
@@ -53,96 +53,3 @@ TEST (election_scheduler, activate_one_flush)
 	system.nodes[0]->scheduler.priority.activate (system.nodes[0]->ledger.tx_begin_read (), nano::dev::genesis_key.pub);
 	ASSERT_TIMELY (5s, system.nodes[0]->active.election (send1->qualified_root ()));
 }
-
-/**
- * Tests that the election scheduler and the active transactions container (AEC)
- * work in sync with regards to the node configuration value "active_elections.size".
- *
- * The test sets up two forcefully cemented blocks -- a send on the genesis account and a receive on a second account.
- * It then creates two other blocks, each a successor to one of the previous two,
- * and processes them locally (without the node starting elections for them, but just saving them to disk).
- *
- * Elections for these latter two (B1 and B2) are started by the test code manually via `election_scheduler::activate`.
- * The test expects E1 to start right off and take its seat into the AEC.
- * E2 is expected not to start though (because the AEC is full), so B2 should be awaiting in the scheduler's queue.
- *
- * As soon as the test code manually confirms E1 (and thus evicts it out of the AEC),
- * it is expected that E2 begins and the scheduler's queue becomes empty again.
- */
-TEST (election_scheduler, no_vacancy)
-{
-	nano::test::system system{};
-
-	nano::node_config config = system.default_config ();
-	config.active_elections.size = 1;
-	config.frontiers_confirmation = nano::frontiers_confirmation_mode::disabled;
-
-	auto & node = *system.add_node (config);
-	nano::state_block_builder builder{};
-	nano::keypair key{};
-
-	// Activating accounts depends on confirmed dependencies. First, prepare 2 accounts
-	auto send = builder.make_block ()
-				.account (nano::dev::genesis_key.pub)
-				.previous (nano::dev::genesis->hash ())
-				.representative (nano::dev::genesis_key.pub)
-				.link (key.pub)
-				.balance (nano::dev::constants.genesis_amount - nano::Gxrb_ratio)
-				.sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
-				.work (*system.work.generate (nano::dev::genesis->hash ()))
-				.build ();
-	ASSERT_EQ (nano::block_status::progress, node.process (send));
-	node.process_confirmed (nano::election_status{ send });
-
-	auto receive = builder.make_block ()
-				   .account (key.pub)
-				   .previous (0)
-				   .representative (key.pub)
-				   .link (send->hash ())
-				   .balance (nano::Gxrb_ratio)
-				   .sign (key.prv, key.pub)
-				   .work (*system.work.generate (key.pub))
-				   .build ();
-	ASSERT_EQ (nano::block_status::progress, node.process (receive));
-	node.process_confirmed (nano::election_status{ receive });
-
-	ASSERT_TIMELY (5s, nano::test::confirmed (node, { send, receive }));
-
-	// Second, process two eligible transactions
-	auto block1 = builder.make_block ()
-				  .account (nano::dev::genesis_key.pub)
-				  .previous (send->hash ())
-				  .representative (nano::dev::genesis_key.pub)
-				  .link (nano::dev::genesis_key.pub)
-				  .balance (nano::dev::constants.genesis_amount - 2 * nano::Gxrb_ratio)
-				  .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
-				  .work (*system.work.generate (send->hash ()))
-				  .build ();
-	ASSERT_EQ (nano::block_status::progress, node.process (block1));
-
-	// There is vacancy so it should be inserted
-	node.scheduler.priority.activate (node.ledger.tx_begin_read (), nano::dev::genesis_key.pub);
-	std::shared_ptr<nano::election> election{};
-	ASSERT_TIMELY (5s, (election = node.active.election (block1->qualified_root ())) != nullptr);
-
-	auto block2 = builder.make_block ()
-				  .account (key.pub)
-				  .previous (receive->hash ())
-				  .representative (key.pub)
-				  .link (key.pub)
-				  .balance (0)
-				  .sign (key.prv, key.pub)
-				  .work (*system.work.generate (receive->hash ()))
-				  .build ();
-	ASSERT_EQ (nano::block_status::progress, node.process (block2));
-
-	// There is no vacancy so it should stay queued
-	node.scheduler.priority.activate (node.ledger.tx_begin_read (), key.pub);
-	ASSERT_TIMELY_EQ (5s, node.scheduler.priority.size (), 1);
-	ASSERT_EQ (node.active.election (block2->qualified_root ()), nullptr);
-
-	// Election confirmed, next in queue should begin
-	election->force_confirm ();
-	ASSERT_TIMELY (5s, node.active.election (block2->qualified_root ()) != nullptr);
-	ASSERT_TRUE (node.scheduler.priority.empty ());
-}
diff --git a/nano/core_test/scheduler_buckets.cpp b/nano/core_test/scheduler_buckets.cpp
index 43912e29b9..47c18ec2ed 100644
--- a/nano/core_test/scheduler_buckets.cpp
+++ b/nano/core_test/scheduler_buckets.cpp
@@ -1,5 +1,5 @@
 #include <nano/lib/blocks.hpp>
-#include <nano/node/scheduler/buckets.hpp>
+#include <nano/node/scheduler/bucket.hpp>
 #include <nano/secure/common.hpp>
 
 #include <gtest/gtest.h>
@@ -107,148 +107,50 @@ std::shared_ptr<nano::state_block> & block3 ()
 	return result;
 }
 
-TEST (buckets, construction)
+TEST (bucket, construction)
 {
-	nano::scheduler::buckets buckets;
-	ASSERT_EQ (0, buckets.size ());
-	ASSERT_TRUE (buckets.empty ());
-	ASSERT_EQ (62, buckets.bucket_count ());
+	nano::scheduler::bucket bucket{ 0 };
 }
 
-TEST (buckets, index_min)
+TEST (bucket, insert_zero)
 {
-	nano::scheduler::buckets buckets;
-	ASSERT_EQ (0, buckets.index (std::numeric_limits<nano::uint128_t>::min ()));
+	nano::scheduler::bucket bucket{ 0 };
+	auto block = block0 ();
+	auto drop = bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 0 }, block);
+	ASSERT_EQ (drop, block);
 }
 
-TEST (buckets, index_max)
+TEST (bucket, push_available)
 {
-	nano::scheduler::buckets buckets;
-	ASSERT_EQ (buckets.bucket_count () - 1, buckets.index (std::numeric_limits<nano::uint128_t>::max ()));
+	nano::scheduler::bucket bucket{ 1 };
+	auto block = block0 ();
+	ASSERT_EQ (nullptr, bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 1000 }, block));
 }
 
-TEST (buckets, insert_Gxrb)
+TEST (bucket, push_overflow_other)
 {
-	nano::scheduler::buckets buckets;
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (1, buckets.bucket_size (48));
+	nano::scheduler::bucket bucket{ 1 };
+	auto block0 = ::block0 ();
+	auto block1 = ::block1 ();
+	ASSERT_EQ (nullptr, bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 1000 }, block0));
+	ASSERT_EQ (block0, bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 900 }, block1));
 }
 
-TEST (buckets, insert_Mxrb)
+TEST (bucket, push_overflow_self)
 {
-	nano::scheduler::buckets buckets;
-	buckets.push (1000, block1 (), nano::Mxrb_ratio);
-	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (1, buckets.bucket_size (13));
+	nano::scheduler::bucket bucket{ 1 };
+	auto block0 = ::block0 ();
+	auto block1 = ::block1 ();
+	ASSERT_EQ (nullptr, bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 1000 }, block0));
+	ASSERT_EQ (block1, bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 1100 }, block1));
 }
 
-// Test two blocks with the same priority
-TEST (buckets, insert_same_priority)
+// Inserting duplicate block should not return an overflow or reject
+TEST (bucket, accept_duplicate)
 {
-	nano::scheduler::buckets buckets;
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	buckets.push (1000, block2 (), nano::Gxrb_ratio);
-	ASSERT_EQ (2, buckets.size ());
-	ASSERT_EQ (2, buckets.bucket_size (48));
-}
-
-// Test the same block inserted multiple times
-TEST (buckets, insert_duplicate)
-{
-	nano::scheduler::buckets buckets;
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (1, buckets.bucket_size (48));
-}
-
-TEST (buckets, insert_older)
-{
-	nano::scheduler::buckets buckets;
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	buckets.push (1100, block2 (), nano::Gxrb_ratio);
-	ASSERT_EQ (block0 (), buckets.top ());
-	buckets.pop ();
-	ASSERT_EQ (block2 (), buckets.top ());
-	buckets.pop ();
-}
-
-TEST (buckets, pop)
-{
-	nano::scheduler::buckets buckets;
-	ASSERT_TRUE (buckets.empty ());
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	ASSERT_FALSE (buckets.empty ());
-	buckets.pop ();
-	ASSERT_TRUE (buckets.empty ());
-}
-
-TEST (buckets, top_one)
-{
-	nano::scheduler::buckets buckets;
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	ASSERT_EQ (block0 (), buckets.top ());
-}
-
-TEST (buckets, top_two)
-{
-	nano::scheduler::buckets buckets;
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	buckets.push (1, block1 (), nano::Mxrb_ratio);
-	ASSERT_EQ (block0 (), buckets.top ());
-	buckets.pop ();
-	ASSERT_EQ (block1 (), buckets.top ());
-	buckets.pop ();
-	ASSERT_TRUE (buckets.empty ());
-}
-
-TEST (buckets, top_round_robin)
-{
-	nano::scheduler::buckets buckets;
-	buckets.push (1000, blockzero (), 0);
-	ASSERT_EQ (blockzero (), buckets.top ());
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	buckets.push (1000, block1 (), nano::Mxrb_ratio);
-	buckets.push (1100, block3 (), nano::Mxrb_ratio);
-	buckets.pop (); // blockzero
-	EXPECT_EQ (block1 (), buckets.top ());
-	buckets.pop ();
-	EXPECT_EQ (block0 (), buckets.top ());
-	buckets.pop ();
-	EXPECT_EQ (block3 (), buckets.top ());
-	buckets.pop ();
-	EXPECT_TRUE (buckets.empty ());
-}
-
-TEST (buckets, trim_normal)
-{
-	nano::scheduler::buckets buckets{ 1 };
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	buckets.push (1100, block2 (), nano::Gxrb_ratio);
-	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (block0 (), buckets.top ());
-}
-
-TEST (buckets, trim_reverse)
-{
-	nano::scheduler::buckets buckets{ 1 };
-	buckets.push (1100, block2 (), nano::Gxrb_ratio);
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (block0 (), buckets.top ());
-}
-
-TEST (buckets, trim_even)
-{
-	nano::scheduler::buckets buckets{ 2 };
-	buckets.push (1000, block0 (), nano::Gxrb_ratio);
-	buckets.push (1100, block2 (), nano::Gxrb_ratio);
-	ASSERT_EQ (1, buckets.size ());
-	ASSERT_EQ (block0 (), buckets.top ());
-	buckets.push (1000, block1 (), nano::Mxrb_ratio);
-	ASSERT_EQ (2, buckets.size ());
-	ASSERT_EQ (block0 (), buckets.top ());
-	buckets.pop ();
-	ASSERT_EQ (block1 (), buckets.top ());
+	nano::scheduler::bucket bucket{ 1 };
+	auto block0 = ::block0 ();
+	ASSERT_EQ (nullptr, bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 1000 }, block0));
+	ASSERT_EQ (nullptr, bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 900 }, block0));
+	ASSERT_EQ (nullptr, bucket.insert (std::chrono::steady_clock::time_point{} + std::chrono::milliseconds{ 1100 }, block0));
 }
diff --git a/nano/lib/logging_enums.hpp b/nano/lib/logging_enums.hpp
index 662981d0a6..94e7715910 100644
--- a/nano/lib/logging_enums.hpp
+++ b/nano/lib/logging_enums.hpp
@@ -126,6 +126,9 @@ enum class detail
 
 	// election_scheduler
 	block_activated,
+	block_insert, // Block inserted with no overflow
+	block_overflow, // Block inserted and another block was removed as an overflow
+	block_reject, // Block was not inserted because of an overflow
 
 	// vote_generator
 	candidate_processed,
@@ -215,4 +218,4 @@ struct magic_enum::customize::enum_range<nano::log::detail>
 {
 	static constexpr int min = 0;
 	static constexpr int max = 512;
-};
\ No newline at end of file
+};
diff --git a/nano/lib/stats_enums.hpp b/nano/lib/stats_enums.hpp
index 616a8a573e..8ec1020b6e 100644
--- a/nano/lib/stats_enums.hpp
+++ b/nano/lib/stats_enums.hpp
@@ -145,6 +145,7 @@ enum class detail
 	balance_mismatch,
 	representative_mismatch,
 	block_position,
+	overflow,
 
 	// blockprocessor
 	process_blocking,
@@ -219,7 +220,8 @@ enum class detail
 	broadcast_block_repeat,
 
 	// election types
-	normal,
+	manual,
+	priority,
 	hinted,
 	optimistic,
 
@@ -299,6 +301,7 @@ enum class detail
 	blocks_confirmed,
 	blocks_confirmed_unbounded,
 	blocks_confirmed_bounded,
+	block_late_removed,
 
 	// request aggregator
 	aggregator_accepted,
@@ -369,6 +372,9 @@ enum class detail
 	insert_priority,
 	insert_priority_success,
 	erase_oldest,
+	block_insert,
+	block_overflow,
+	block_reject,
 
 	// handshake
 	invalid_node_id,
diff --git a/nano/lib/thread_roles.cpp b/nano/lib/thread_roles.cpp
index 8b1b7618e0..5ba7be622d 100644
--- a/nano/lib/thread_roles.cpp
+++ b/nano/lib/thread_roles.cpp
@@ -103,9 +103,6 @@ std::string nano::thread_role::get_string (nano::thread_role::name role)
 		case nano::thread_role::name::scheduler_optimistic:
 			thread_role_name_string = "Sched Opt";
 			break;
-		case nano::thread_role::name::scheduler_priority:
-			thread_role_name_string = "Sched Priority";
-			break;
 		case nano::thread_role::name::stats:
 			thread_role_name_string = "Stats";
 			break;
diff --git a/nano/lib/thread_roles.hpp b/nano/lib/thread_roles.hpp
index ceaeac6652..8949485730 100644
--- a/nano/lib/thread_roles.hpp
+++ b/nano/lib/thread_roles.hpp
@@ -41,7 +41,6 @@ enum class name
 	scheduler_hinted,
 	scheduler_manual,
 	scheduler_optimistic,
-	scheduler_priority,
 	rep_crawler,
 	local_block_broadcasting,
 	rep_tiers,
diff --git a/nano/node/CMakeLists.txt b/nano/node/CMakeLists.txt
index a069e9715e..da45aadffb 100644
--- a/nano/node/CMakeLists.txt
+++ b/nano/node/CMakeLists.txt
@@ -139,8 +139,6 @@ add_library(
   request_aggregator.cpp
   scheduler/bucket.cpp
   scheduler/bucket.hpp
-  scheduler/buckets.cpp
-  scheduler/buckets.hpp
   scheduler/component.hpp
   scheduler/component.cpp
   scheduler/hinted.hpp
diff --git a/nano/node/active_elections.cpp b/nano/node/active_elections.cpp
index 3c223752df..bbf0a7e6ee 100644
--- a/nano/node/active_elections.cpp
+++ b/nano/node/active_elections.cpp
@@ -187,7 +187,11 @@ int64_t nano::active_elections::limit (nano::election_behavior behavior) const
 {
 	switch (behavior)
 	{
-		case nano::election_behavior::normal:
+		case nano::election_behavior::manual:
+		{
+			return std::numeric_limits<int64_t>::max ();
+		}
+		case nano::election_behavior::priority:
 		{
 			return static_cast<int64_t> (config.size);
 		}
@@ -212,8 +216,10 @@ int64_t nano::active_elections::vacancy (nano::election_behavior behavior) const
 	nano::lock_guard<nano::mutex> guard{ mutex };
 	switch (behavior)
 	{
-		case nano::election_behavior::normal:
-			return limit () - static_cast<int64_t> (roots.size ());
+		case nano::election_behavior::manual:
+			return std::numeric_limits<int64_t>::max ();
+		case nano::election_behavior::priority:
+			return limit (nano::election_behavior::priority) - static_cast<int64_t> (roots.size ());
 		case nano::election_behavior::hinted:
 		case nano::election_behavior::optimistic:
 			return limit (behavior) - count_by_behavior[behavior];
@@ -286,7 +292,7 @@ void nano::active_elections::cleanup_election (nano::unique_lock<nano::mutex> &
 
 	node.stats.sample (nano::stat::sample::active_election_duration, { 0, 1000 * 60 * 10 /* 0-10 minutes range */ }, election->duration ().count ());
 
-	vacancy_update ();
+	election_stopped.notify (election);
 
 	for (auto const & [hash, block] : blocks_l)
 	{
@@ -359,20 +365,6 @@ void nano::active_elections::request_loop ()
 	}
 }
 
-void nano::active_elections::trim ()
-{
-	/*
-	 * Both normal and hinted election schedulers are well-behaved, meaning they first check for AEC vacancy before inserting new elections.
-	 * However, it is possible that AEC will be temporarily overfilled in case it's running at full capacity and election hinting or manual queue kicks in.
-	 * That case will lead to unwanted churning of elections, so this allows for AEC to be overfilled to 125% until erasing of elections happens.
-	 */
-	while (vacancy () < -(limit () / 4))
-	{
-		node.stats.inc (nano::stat::type::active, nano::stat::detail::erase_oldest);
-		erase_oldest ();
-	}
-}
-
 nano::election_insertion_result nano::active_elections::insert (std::shared_ptr<nano::block> const & block_a, nano::election_behavior election_behavior_a)
 {
 	debug_assert (block_a);
@@ -434,7 +426,6 @@ nano::election_insertion_result nano::active_elections::insert (std::shared_ptr<
 
 		node.vote_router.trigger_vote_cache (hash);
 		node.observers.active_started.notify (hash);
-		vacancy_update ();
 	}
 
 	// Votes are generated for inserted or ongoing elections
@@ -443,8 +434,6 @@ nano::election_insertion_result nano::active_elections::insert (std::shared_ptr<
 		result.election->broadcast_vote ();
 	}
 
-	trim ();
-
 	return result;
 }
 
@@ -489,16 +478,6 @@ bool nano::active_elections::erase (nano::qualified_root const & root_a)
 	return false;
 }
 
-void nano::active_elections::erase_oldest ()
-{
-	nano::unique_lock<nano::mutex> lock{ mutex };
-	if (!roots.empty ())
-	{
-		auto item = roots.get<tag_sequenced> ().front ();
-		cleanup_election (lock, item.election);
-	}
-}
-
 bool nano::active_elections::empty () const
 {
 	nano::lock_guard<nano::mutex> lock{ mutex };
@@ -547,7 +526,6 @@ void nano::active_elections::clear ()
 		nano::lock_guard<nano::mutex> guard{ mutex };
 		roots.clear ();
 	}
-	vacancy_update ();
 }
 
 std::unique_ptr<nano::container_info_component> nano::collect_container_info (active_elections & active_elections, std::string const & name)
@@ -557,7 +535,7 @@ std::unique_ptr<nano::container_info_component> nano::collect_container_info (ac
 	auto composite = std::make_unique<container_info_composite> (name);
 	composite->add_component (std::make_unique<container_info_leaf> (container_info{ "roots", active_elections.roots.size (), sizeof (decltype (active_elections.roots)::value_type) }));
 	composite->add_component (std::make_unique<container_info_leaf> (container_info{ "election_winner_details", active_elections.election_winner_details_size (), sizeof (decltype (active_elections.election_winner_details)::value_type) }));
-	composite->add_component (std::make_unique<container_info_leaf> (container_info{ "normal", static_cast<std::size_t> (active_elections.count_by_behavior[nano::election_behavior::normal]), 0 }));
+	composite->add_component (std::make_unique<container_info_leaf> (container_info{ "normal", static_cast<std::size_t> (active_elections.count_by_behavior[nano::election_behavior::priority]), 0 }));
 	composite->add_component (std::make_unique<container_info_leaf> (container_info{ "hinted", static_cast<std::size_t> (active_elections.count_by_behavior[nano::election_behavior::hinted]), 0 }));
 	composite->add_component (std::make_unique<container_info_leaf> (container_info{ "optimistic", static_cast<std::size_t> (active_elections.count_by_behavior[nano::election_behavior::optimistic]), 0 }));
 
diff --git a/nano/node/active_elections.hpp b/nano/node/active_elections.hpp
index 5631b8c2f8..ddcb2645c8 100644
--- a/nano/node/active_elections.hpp
+++ b/nano/node/active_elections.hpp
@@ -108,7 +108,7 @@ class active_elections final
 	/**
 	 * Starts new election with a specified behavior type
 	 */
-	nano::election_insertion_result insert (std::shared_ptr<nano::block> const &, nano::election_behavior = nano::election_behavior::normal);
+	nano::election_insertion_result insert (std::shared_ptr<nano::block> const &, nano::election_behavior = nano::election_behavior::priority);
 	// Is the root of this block in the roots container
 	bool active (nano::block const &) const;
 	bool active (nano::qualified_root const &) const;
@@ -117,7 +117,6 @@ class active_elections final
 	std::vector<std::shared_ptr<nano::election>> list_active (std::size_t = std::numeric_limits<std::size_t>::max ());
 	bool erase (nano::block const &);
 	bool erase (nano::qualified_root const &);
-	void erase_oldest ();
 	bool empty () const;
 	std::size_t size () const;
 	bool publish (std::shared_ptr<nano::block> const &);
@@ -128,20 +127,18 @@ class active_elections final
 	 * Maximum number of elections that should be present in this container
 	 * NOTE: This is only a soft limit, it is possible for this container to exceed this count
 	 */
-	int64_t limit (nano::election_behavior behavior = nano::election_behavior::normal) const;
+	int64_t limit (nano::election_behavior behavior) const;
 	/**
 	 * How many election slots are available for specified election type
 	 */
-	int64_t vacancy (nano::election_behavior behavior = nano::election_behavior::normal) const;
-	std::function<void ()> vacancy_update{ [] () {} };
+	int64_t vacancy (nano::election_behavior behavior) const;
+	nano::observer_set<std::shared_ptr<nano::election>> election_stopped;
 
 	std::size_t election_winner_details_size ();
 	void add_election_winner_details (nano::block_hash const &, std::shared_ptr<nano::election> const &);
 	std::shared_ptr<nano::election> remove_election_winner_details (nano::block_hash const &);
 
 private:
-	// Erase elections if we're over capacity
-	void trim ();
 	void request_loop ();
 	void request_confirm (nano::unique_lock<nano::mutex> &);
 	// Erase all blocks from active and, if not confirmed, clear digests from network filters
diff --git a/nano/node/blockprocessor.cpp b/nano/node/blockprocessor.cpp
index 80ecefeb62..3b0a7b7625 100644
--- a/nano/node/blockprocessor.cpp
+++ b/nano/node/blockprocessor.cpp
@@ -6,6 +6,8 @@
 #include <nano/node/blockprocessor.hpp>
 #include <nano/node/local_vote_history.hpp>
 #include <nano/node/node.hpp>
+#include <nano/node/scheduler/component.hpp>
+#include <nano/node/scheduler/priority.hpp>
 #include <nano/secure/ledger.hpp>
 #include <nano/secure/ledger_set_any.hpp>
 #include <nano/store/component.hpp>
@@ -367,6 +369,22 @@ nano::block_status nano::block_processor::process_one (secure::write_transaction
 	nano::log::arg{ "forced", forced_a },
 	nano::log::arg{ "block", block });
 
+	if (result == nano::block_status::progress && context.source == nano::block_source::live)
+	{
+		std::shared_ptr<nano::block> removed;
+		if (node.ledger.dependents_confirmed (transaction_a, *block))
+		{
+			removed = node.scheduler.priority.activate (transaction_a, block->account ());
+		}
+		if (removed != nullptr && node.ledger.any.block_exists (transaction_a, removed->hash ()))
+		{
+			//node.ledger.rollback (transaction_a, removed->hash ());
+			if (removed->hash () == block->hash ())
+			{
+				return nano::block_status::overflow;
+			}
+		}
+	}
 	switch (result)
 	{
 		case nano::block_status::progress:
@@ -444,6 +462,10 @@ nano::block_status nano::block_processor::process_one (secure::write_transaction
 		{
 			break;
 		}
+		case nano::block_status::overflow:
+		{
+			break;
+		}
 	}
 	return result;
 }
diff --git a/nano/node/confirming_set.cpp b/nano/node/confirming_set.cpp
index d39cf5791d..6161714a04 100644
--- a/nano/node/confirming_set.cpp
+++ b/nano/node/confirming_set.cpp
@@ -1,6 +1,7 @@
 #include <nano/lib/thread_roles.hpp>
 #include <nano/node/confirming_set.hpp>
 #include <nano/secure/ledger.hpp>
+#include <nano/secure/ledger_set_any.hpp>
 #include <nano/secure/ledger_set_confirmed.hpp>
 #include <nano/store/component.hpp>
 #include <nano/store/write_queue.hpp>
@@ -80,15 +81,22 @@ void nano::confirming_set::run ()
 				{
 					auto item = *i;
 					lock.unlock ();
-					auto added = ledger.confirm (tx, item);
-					if (!added.empty ())
+					if (ledger.any.block_exists (tx, item))
 					{
-						// Confirming this block may implicitly confirm more
-						cemented.insert (cemented.end (), added.begin (), added.end ());
+						auto added = ledger.confirm (tx, item);
+						if (!added.empty ())
+						{
+							// Confirming this block may implicitly confirm more
+							cemented.insert (cemented.end (), added.begin (), added.end ());
+						}
+						else
+						{
+							already.push_back (item);
+						}
 					}
 					else
 					{
-						already.push_back (item);
+						ledger.stats.inc (nano::stat::type::confirmation_height, nano::stat::detail::block_late_removed);
 					}
 					lock.lock ();
 				}
diff --git a/nano/node/election.cpp b/nano/node/election.cpp
index 56415165a1..02d6a0cb1a 100644
--- a/nano/node/election.cpp
+++ b/nano/node/election.cpp
@@ -142,7 +142,8 @@ std::chrono::milliseconds nano::election::confirm_req_time () const
 {
 	switch (behavior ())
 	{
-		case election_behavior::normal:
+		case election_behavior::manual:
+		case election_behavior::priority:
 		case election_behavior::hinted:
 			return base_latency () * 5;
 		case election_behavior::optimistic:
@@ -295,7 +296,8 @@ std::chrono::milliseconds nano::election::time_to_live () const
 {
 	switch (behavior ())
 	{
-		case election_behavior::normal:
+		case election_behavior::manual:
+		case election_behavior::priority:
 			return std::chrono::milliseconds (5 * 60 * 1000);
 		case election_behavior::hinted:
 		case election_behavior::optimistic:
diff --git a/nano/node/election.hpp b/nano/node/election.hpp
index 2ebe729b8f..6111191e2b 100644
--- a/nano/node/election.hpp
+++ b/nano/node/election.hpp
@@ -3,7 +3,6 @@
 #include <nano/lib/id_dispenser.hpp>
 #include <nano/lib/logging.hpp>
 #include <nano/lib/stats_enums.hpp>
-#include <nano/node/election_behavior.hpp>
 #include <nano/node/election_status.hpp>
 #include <nano/node/vote_with_weight_info.hpp>
 #include <nano/secure/common.hpp>
@@ -17,6 +16,7 @@ namespace nano
 class block;
 class channel;
 class confirmation_solicitor;
+enum class election_behavior;
 class inactive_cache_information;
 class node;
 enum class vote_code;
@@ -172,7 +172,7 @@ class election final : public std::enable_shared_from_this<election>
 	mutable nano::uint128_t final_weight{ 0 };
 	mutable std::unordered_map<nano::block_hash, nano::uint128_t> last_tally;
 
-	nano::election_behavior const behavior_m{ nano::election_behavior::normal };
+	nano::election_behavior const behavior_m;
 	std::chrono::steady_clock::time_point const election_start{ std::chrono::steady_clock::now () };
 
 	mutable nano::mutex mutex;
diff --git a/nano/node/election_behavior.hpp b/nano/node/election_behavior.hpp
index c38b1c8de6..1224b1787d 100644
--- a/nano/node/election_behavior.hpp
+++ b/nano/node/election_behavior.hpp
@@ -8,7 +8,8 @@ namespace nano
 {
 enum class election_behavior
 {
-	normal,
+	manual,
+	priority,
 	/**
 	 * Hinted elections:
 	 * - shorter timespan
diff --git a/nano/node/json_handler.cpp b/nano/node/json_handler.cpp
index d1978048ef..62a5ee193c 100644
--- a/nano/node/json_handler.cpp
+++ b/nano/node/json_handler.cpp
@@ -2009,7 +2009,8 @@ void nano::json_handler::confirmation_active ()
 void nano::json_handler::election_statistics ()
 {
 	auto active_elections = node.active.list_active ();
-	unsigned normal_count = 0;
+	unsigned manual_count = 0;
+	unsigned priority_count = 0;
 	unsigned hinted_count = 0;
 	unsigned optimistic_count = 0;
 	unsigned total_count = 0;
@@ -2027,8 +2028,11 @@ void nano::json_handler::election_statistics ()
 
 		switch (election->behavior ())
 		{
-			case election_behavior::normal:
-				normal_count++;
+			case election_behavior::manual:
+				manual_count++;
+				break;
+			case election_behavior::priority:
+				priority_count++;
 				break;
 			case election_behavior::hinted:
 				hinted_count++;
@@ -2046,7 +2050,7 @@ void nano::json_handler::election_statistics ()
 	std::stringstream stream_utilization;
 	stream_utilization << std::fixed << std::setprecision (2) << utilization_percentage;
 
-	response_l.put ("normal", normal_count);
+	response_l.put ("priority", priority_count);
 	response_l.put ("hinted", hinted_count);
 	response_l.put ("optimistic", optimistic_count);
 	response_l.put ("total", total_count);
diff --git a/nano/node/node.cpp b/nano/node/node.cpp
index 40040c18ad..9dbaeb4807 100644
--- a/nano/node/node.cpp
+++ b/nano/node/node.cpp
@@ -253,11 +253,11 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
 	if (!init_error ())
 	{
 		// Notify election schedulers when AEC frees election slot
-		active.vacancy_update = [this] () {
-			scheduler.priority.notify ();
+		active.election_stopped.add ([this] (std::shared_ptr<nano::election> election) {
+			scheduler.priority.election_stopped (election);
 			scheduler.hinted.notify ();
 			scheduler.optimistic.notify ();
-		};
+		});
 
 		wallets.observer = [this] (bool active) {
 			observers.wallet.notify (active);
diff --git a/nano/node/nodeconfig.hpp b/nano/node/nodeconfig.hpp
index a17cecacd5..4b00e5ed6e 100644
--- a/nano/node/nodeconfig.hpp
+++ b/nano/node/nodeconfig.hpp
@@ -18,6 +18,7 @@
 #include <nano/node/request_aggregator.hpp>
 #include <nano/node/scheduler/hinted.hpp>
 #include <nano/node/scheduler/optimistic.hpp>
+#include <nano/node/scheduler/priority.hpp>
 #include <nano/node/transport/tcp_listener.hpp>
 #include <nano/node/vote_cache.hpp>
 #include <nano/node/vote_processor.hpp>
@@ -61,6 +62,7 @@ class node_config
 	std::optional<uint16_t> peering_port{};
 	nano::scheduler::optimistic_config optimistic_scheduler;
 	nano::scheduler::hinted_config hinted_scheduler;
+	nano::scheduler::priority_config priority_scheduler;
 	std::vector<std::pair<std::string, uint16_t>> work_peers;
 	std::vector<std::pair<std::string, uint16_t>> secondary_work_peers{ { "127.0.0.1", 8076 } }; /* Default of nano-pow-server */
 	std::vector<std::string> preconfigured_peers;
diff --git a/nano/node/process_live_dispatcher.cpp b/nano/node/process_live_dispatcher.cpp
index 60a80e2154..15a051eff8 100644
--- a/nano/node/process_live_dispatcher.cpp
+++ b/nano/node/process_live_dispatcher.cpp
@@ -44,10 +44,10 @@ void nano::process_live_dispatcher::inspect (nano::block_status const & result,
 void nano::process_live_dispatcher::process_live (nano::block const & block, secure::transaction const & transaction)
 {
 	// Start collecting quorum on block
-	if (ledger.dependents_confirmed (transaction, block))
+	/*if (ledger.dependents_confirmed (transaction, block))
 	{
 		scheduler.activate (transaction, block.account ());
-	}
+	}*/
 
 	if (websocket.server && websocket.server->any_subscriber (nano::websocket::topic::new_unconfirmed_block))
 	{
diff --git a/nano/node/scheduler/bucket.cpp b/nano/node/scheduler/bucket.cpp
index 51ba1626a5..b435d5c3aa 100644
--- a/nano/node/scheduler/bucket.cpp
+++ b/nano/node/scheduler/bucket.cpp
@@ -1,62 +1,34 @@
 #include <nano/lib/blocks.hpp>
 #include <nano/node/scheduler/bucket.hpp>
 
-bool nano::scheduler::bucket::value_type::operator< (value_type const & other_a) const
+nano::block_hash nano::scheduler::bucket::entry_t::hash () const
 {
-	return time < other_a.time || (time == other_a.time && block->hash () < other_a.block->hash ());
+	return block->hash ();
 }
 
-bool nano::scheduler::bucket::value_type::operator== (value_type const & other_a) const
+nano::scheduler::bucket::bucket (size_t max) :
+	max{ max }
 {
-	return time == other_a.time && block->hash () == other_a.block->hash ();
 }
 
-nano::scheduler::bucket::bucket (size_t maximum) :
-	maximum{ maximum }
+std::shared_ptr<nano::block> nano::scheduler::bucket::insert (std::chrono::steady_clock::time_point time, std::shared_ptr<nano::block> block)
 {
-	debug_assert (maximum > 0);
-}
-
-nano::scheduler::bucket::~bucket ()
-{
-}
-
-std::shared_ptr<nano::block> nano::scheduler::bucket::top () const
-{
-	debug_assert (!queue.empty ());
-	return queue.begin ()->block;
-}
-
-void nano::scheduler::bucket::pop ()
-{
-	debug_assert (!queue.empty ());
-	queue.erase (queue.begin ());
-}
-
-void nano::scheduler::bucket::push (uint64_t time, std::shared_ptr<nano::block> block)
-{
-	queue.insert ({ time, block });
-	if (queue.size () > maximum)
+	std::lock_guard lock{ mutex };
+	backlog.insert (entry_t{ time, block });
+	debug_assert (backlog.size () <= max + 1); // One extra at most
+	if (backlog.size () <= max)
 	{
-		debug_assert (!queue.empty ());
-		queue.erase (--queue.end ());
+		return nullptr;
 	}
+	auto newest = backlog.begin (); // The first item in descending order has the highest timestamp i.e. it is the newest.
+	auto discard = newest->block;
+	backlog.erase (newest);
+	debug_assert (backlog.size () <= max);
+	return discard;
 }
 
-size_t nano::scheduler::bucket::size () const
-{
-	return queue.size ();
-}
-
-bool nano::scheduler::bucket::empty () const
+size_t nano::scheduler::bucket::erase (nano::block_hash const & hash)
 {
-	return queue.empty ();
-}
-
-void nano::scheduler::bucket::dump () const
-{
-	for (auto const & item : queue)
-	{
-		std::cerr << item.time << ' ' << item.block->hash ().to_string () << '\n';
-	}
+	std::lock_guard lock{ mutex };
+	return backlog.get<tag_hash> ().erase (hash);
 }
diff --git a/nano/node/scheduler/bucket.hpp b/nano/node/scheduler/bucket.hpp
index 2f32c17d59..c2fade950a 100644
--- a/nano/node/scheduler/bucket.hpp
+++ b/nano/node/scheduler/bucket.hpp
@@ -1,13 +1,24 @@
 #pragma once
 
+#include <nano/lib/timer.hpp>
+#include <nano/secure/common.hpp>
+
+#include <boost/multi_index/hashed_index.hpp>
+#include <boost/multi_index/mem_fun.hpp>
+#include <boost/multi_index/member.hpp>
+#include <boost/multi_index/ordered_index.hpp>
+#include <boost/multi_index_container.hpp>
+
+#include <chrono>
 #include <cstddef>
 #include <cstdint>
 #include <memory>
-#include <set>
+#include <mutex>
 
 namespace nano
 {
 class block;
+class logger;
 }
 namespace nano::scheduler
 {
@@ -15,25 +26,35 @@ namespace nano::scheduler
  */
 class bucket final
 {
-	class value_type
+	class tag_time
+	{
+	};
+	class tag_hash
+	{
+	};
+	class entry_t
 	{
 	public:
-		uint64_t time;
+		std::chrono::steady_clock::time_point time;
 		std::shared_ptr<nano::block> block;
-		bool operator< (value_type const & other_a) const;
-		bool operator== (value_type const & other_a) const;
+		nano::block_hash hash () const;
 	};
-	std::set<value_type> queue;
-	size_t const maximum;
+	using backlog_t = boost::multi_index_container<entry_t,
+	boost::multi_index::indexed_by<
+	boost::multi_index::ordered_non_unique<boost::multi_index::tag<tag_time>,
+	boost::multi_index::member<entry_t, std::chrono::steady_clock::time_point, &entry_t::time>,
+	std::greater<std::chrono::steady_clock::time_point>>, // Sorted by last-confirmed time in descending order
+	boost::multi_index::hashed_unique<boost::multi_index::tag<tag_hash>,
+	boost::multi_index::const_mem_fun<entry_t, nano::block_hash, &entry_t::hash>>>>;
 
 public:
-	bucket (size_t maximum);
-	~bucket ();
-	std::shared_ptr<nano::block> top () const;
-	void pop ();
-	void push (uint64_t time, std::shared_ptr<nano::block> block);
-	size_t size () const;
-	bool empty () const;
-	void dump () const;
+	bucket (size_t max);
+	std::shared_ptr<nano::block> insert (std::chrono::steady_clock::time_point time, std::shared_ptr<nano::block> block);
+	size_t erase (nano::block_hash const & hash);
+
+private:
+	backlog_t backlog;
+	std::mutex mutex;
+	size_t const max;
 };
 } // namespace nano::scheduler
diff --git a/nano/node/scheduler/buckets.cpp b/nano/node/scheduler/buckets.cpp
deleted file mode 100644
index c033f6f6ca..0000000000
--- a/nano/node/scheduler/buckets.cpp
+++ /dev/null
@@ -1,150 +0,0 @@
-#include <nano/lib/blocks.hpp>
-#include <nano/lib/utility.hpp>
-#include <nano/node/scheduler/bucket.hpp>
-#include <nano/node/scheduler/buckets.hpp>
-
-#include <string>
-
-/** Moves the bucket pointer to the next bucket */
-void nano::scheduler::buckets::next ()
-{
-	++current;
-	if (current == buckets_m.end ())
-	{
-		current = buckets_m.begin ();
-	}
-}
-
-/** Seek to the next non-empty bucket, if one exists */
-void nano::scheduler::buckets::seek ()
-{
-	next ();
-	for (std::size_t i = 0, n = buckets_m.size (); (*current)->empty () && i < n; ++i)
-	{
-		next ();
-	}
-}
-
-/**
- * Prioritization constructor, construct a container containing approximately 'maximum' number of blocks.
- * @param maximum number of blocks that this container can hold, this is a soft and approximate limit.
- */
-nano::scheduler::buckets::buckets (uint64_t maximum) :
-	maximum{ maximum }
-{
-	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 } << 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);
-	auto bucket_max = std::max<size_t> (1u, maximum / minimums.size ());
-	for (size_t i = 0u, n = minimums.size (); i < n; ++i)
-	{
-		buckets_m.push_back (std::make_unique<scheduler::bucket> (bucket_max));
-	}
-	current = buckets_m.begin ();
-}
-
-nano::scheduler::buckets::~buckets ()
-{
-}
-
-std::size_t nano::scheduler::buckets::index (nano::uint128_t const & balance) const
-{
-	auto index = std::upper_bound (minimums.begin (), minimums.end (), balance) - minimums.begin () - 1;
-	return index;
-}
-
-/**
- * Push a block and its associated time into the prioritization container.
- * The time is given here because sideband might not exist in the case of state blocks.
- */
-void nano::scheduler::buckets::push (uint64_t time, std::shared_ptr<nano::block> block, nano::amount const & priority)
-{
-	auto was_empty = empty ();
-	auto & bucket = buckets_m[index (priority.number ())];
-	bucket->push (time, block);
-	if (was_empty)
-	{
-		seek ();
-	}
-}
-
-/** Return the highest priority block of the current bucket */
-std::shared_ptr<nano::block> nano::scheduler::buckets::top () const
-{
-	debug_assert (!empty ());
-	auto result = (*current)->top ();
-	return result;
-}
-
-/** Pop the current block from the container and seek to the next block, if it exists */
-void nano::scheduler::buckets::pop ()
-{
-	debug_assert (!empty ());
-	auto & bucket = *current;
-	bucket->pop ();
-	seek ();
-}
-
-/** Returns the total number of blocks in buckets */
-std::size_t nano::scheduler::buckets::size () const
-{
-	std::size_t result{ 0 };
-	for (auto const & bucket : buckets_m)
-	{
-		result += bucket->size ();
-	}
-	return result;
-}
-
-/** Returns number of buckets, 62 by default */
-std::size_t nano::scheduler::buckets::bucket_count () const
-{
-	return buckets_m.size ();
-}
-
-/** Returns number of items in bucket with index 'index' */
-std::size_t nano::scheduler::buckets::bucket_size (std::size_t index) const
-{
-	return buckets_m[index]->size ();
-}
-
-/** Returns true if all buckets are empty */
-bool nano::scheduler::buckets::empty () const
-{
-	return std::all_of (buckets_m.begin (), buckets_m.end (), [] (auto const & bucket) { return bucket->empty (); });
-}
-
-/** Print the state of the class in stderr */
-void nano::scheduler::buckets::dump () const
-{
-	for (auto const & bucket : buckets_m)
-	{
-		bucket->dump ();
-	}
-	std::cerr << "current: " << current - buckets_m.begin () << '\n';
-}
-
-std::unique_ptr<nano::container_info_component> nano::scheduler::buckets::collect_container_info (std::string const & name)
-{
-	auto composite = std::make_unique<container_info_composite> (name);
-	for (auto i = 0; i < buckets_m.size (); ++i)
-	{
-		auto const & bucket = buckets_m[i];
-		composite->add_component (std::make_unique<container_info_leaf> (container_info{ std::to_string (i), bucket->size (), 0 }));
-	}
-	return composite;
-}
diff --git a/nano/node/scheduler/buckets.hpp b/nano/node/scheduler/buckets.hpp
deleted file mode 100644
index 967b4408f7..0000000000
--- a/nano/node/scheduler/buckets.hpp
+++ /dev/null
@@ -1,60 +0,0 @@
-#pragma once
-#include <nano/lib/numbers.hpp>
-#include <nano/lib/utility.hpp>
-
-#include <cstddef>
-#include <cstdint>
-#include <deque>
-#include <memory>
-
-namespace nano
-{
-class block;
-}
-namespace nano::scheduler
-{
-class bucket;
-/** A container for holding blocks and their arrival/creation time.
- *
- *  The container consists of a number of buckets. Each bucket holds an ordered set of 'value_type' items.
- *  The buckets are accessed in a round robin fashion. The index 'current' holds the index of the bucket to access next.
- *  When a block is inserted, the bucket to go into is determined by the account balance and the priority inside that
- *  bucket is determined by its creation/arrival time.
- *
- *  The arrival/creation time is only an approximation and it could even be wildly wrong,
- *  for example, in the event of bootstrapped blocks.
- */
-class buckets final
-{
-	/** container for the buckets to be read in round robin fashion */
-	std::deque<std::unique_ptr<bucket>> buckets_m;
-
-	/** thresholds that define the bands for each bucket, the minimum balance an account must have to enter a bucket,
-	 *  the container writes a block to the lowest indexed bucket that has balance larger than the bucket's minimum value */
-	std::deque<nano::uint128_t> minimums;
-
-	/** index of bucket to read next */
-	decltype (buckets_m)::const_iterator current;
-
-	/** maximum number of blocks in whole container, each bucket's maximum is maximum / bucket_number */
-	uint64_t const maximum;
-
-	void next ();
-	void seek ();
-
-public:
-	buckets (uint64_t maximum = 250000u);
-	~buckets ();
-	void push (uint64_t time, std::shared_ptr<nano::block> block, nano::amount const & priority);
-	std::shared_ptr<nano::block> top () const;
-	void pop ();
-	std::size_t size () const;
-	std::size_t bucket_count () const;
-	std::size_t bucket_size (std::size_t index) const;
-	bool empty () const;
-	void dump () const;
-	std::size_t index (nano::uint128_t const & balance) const;
-
-	std::unique_ptr<nano::container_info_component> collect_container_info (std::string const &);
-};
-} // namespace nano::scheduler
diff --git a/nano/node/scheduler/component.cpp b/nano/node/scheduler/component.cpp
index 0d44042448..57b58a3c02 100644
--- a/nano/node/scheduler/component.cpp
+++ b/nano/node/scheduler/component.cpp
@@ -26,7 +26,6 @@ void nano::scheduler::component::start ()
 	hinted.start ();
 	manual.start ();
 	optimistic.start ();
-	priority.start ();
 }
 
 void nano::scheduler::component::stop ()
@@ -34,7 +33,6 @@ void nano::scheduler::component::stop ()
 	hinted.stop ();
 	manual.stop ();
 	optimistic.stop ();
-	priority.stop ();
 }
 
 std::unique_ptr<nano::container_info_component> nano::scheduler::component::collect_container_info (std::string const & name)
diff --git a/nano/node/scheduler/hinted.hpp b/nano/node/scheduler/hinted.hpp
index e26772bda2..330e894055 100644
--- a/nano/node/scheduler/hinted.hpp
+++ b/nano/node/scheduler/hinted.hpp
@@ -61,7 +61,8 @@ class hinted final
 	void stop ();
 
 	/*
-	 * Notify about changes in AEC vacancy
+	 * Notify when AEC has become more empty
+	 * Notify when scheduler is stopped
 	 */
 	void notify ();
 
diff --git a/nano/node/scheduler/manual.cpp b/nano/node/scheduler/manual.cpp
index 85108e6a82..3235a91c20 100644
--- a/nano/node/scheduler/manual.cpp
+++ b/nano/node/scheduler/manual.cpp
@@ -42,7 +42,7 @@ void nano::scheduler::manual::notify ()
 void nano::scheduler::manual::push (std::shared_ptr<nano::block> const & block_a, boost::optional<nano::uint128_t> const & previous_balance_a)
 {
 	nano::lock_guard<nano::mutex> lock{ mutex };
-	queue.push_back (std::make_tuple (block_a, previous_balance_a, nano::election_behavior::normal));
+	queue.push_back (std::make_tuple (block_a, previous_balance_a, nano::election_behavior::manual));
 	notify ();
 }
 
diff --git a/nano/node/scheduler/manual.hpp b/nano/node/scheduler/manual.hpp
index ee6c060048..648a7e934c 100644
--- a/nano/node/scheduler/manual.hpp
+++ b/nano/node/scheduler/manual.hpp
@@ -1,7 +1,6 @@
 #pragma once
 #include <nano/lib/locks.hpp>
 #include <nano/lib/numbers.hpp>
-#include <nano/node/election_behavior.hpp>
 
 #include <boost/optional.hpp>
 
@@ -12,6 +11,7 @@
 namespace nano
 {
 class block;
+enum class election_behavior;
 class node;
 }
 
diff --git a/nano/node/scheduler/optimistic.hpp b/nano/node/scheduler/optimistic.hpp
index 14318773ab..411c2d4dc5 100644
--- a/nano/node/scheduler/optimistic.hpp
+++ b/nano/node/scheduler/optimistic.hpp
@@ -63,7 +63,8 @@ class optimistic final
 	bool activate (nano::secure::transaction const & transaction, nano::account const & account);
 
 	/**
-	 * Notify about changes in AEC vacancy
+	 * Notify when AEC has become more empty
+	 * Notify when scheduler is stopped
 	 */
 	void notify ();
 
diff --git a/nano/node/scheduler/priority.cpp b/nano/node/scheduler/priority.cpp
index 0a91f8e560..4d776bc4bf 100644
--- a/nano/node/scheduler/priority.cpp
+++ b/nano/node/scheduler/priority.cpp
@@ -2,7 +2,7 @@
 #include <nano/node/active_elections.hpp>
 #include <nano/node/election.hpp>
 #include <nano/node/node.hpp>
-#include <nano/node/scheduler/buckets.hpp>
+#include <nano/node/scheduler/bucket.hpp>
 #include <nano/node/scheduler/priority.hpp>
 #include <nano/secure/ledger.hpp>
 #include <nano/secure/ledger_set_any.hpp>
@@ -10,131 +10,150 @@
 
 nano::scheduler::priority::priority (nano::node & node_a, nano::stats & stats_a) :
 	node{ node_a },
-	stats{ stats_a },
-	buckets{ std::make_unique<scheduler::buckets> () }
+	stats{ stats_a }
 {
+	setup_buckets ();
 }
 
 nano::scheduler::priority::~priority ()
 {
-	// Thread must be stopped before destruction
-	debug_assert (!thread.joinable ());
 }
 
-void nano::scheduler::priority::start ()
-{
-	debug_assert (!thread.joinable ());
-
-	thread = std::thread{ [this] () {
-		nano::thread_role::set (nano::thread_role::name::scheduler_priority);
-		run ();
-	} };
-}
-
-void nano::scheduler::priority::stop ()
-{
-	{
-		nano::lock_guard<nano::mutex> lock{ mutex };
-		stopped = true;
-	}
-	notify ();
-	nano::join_or_pass (thread);
-}
-
-bool nano::scheduler::priority::activate (secure::transaction const & transaction, nano::account const & account)
+std::shared_ptr<nano::block> nano::scheduler::priority::activate (secure::transaction const & transaction, nano::account const & account)
 {
 	debug_assert (!account.is_zero ());
 	auto head = node.ledger.confirmed.account_head (transaction, account);
 	if (node.ledger.any.account_head (transaction, account) == head)
 	{
-		return false;
+		return nullptr;
 	}
 	auto block = node.ledger.any.block_get (transaction, node.ledger.any.block_successor (transaction, { head.is_zero () ? static_cast<nano::uint256_union> (account) : head, head }).value ());
 	if (!node.ledger.dependents_confirmed (transaction, *block))
 	{
-		return false;
+		return nullptr;
 	}
-	auto const balance_priority = std::max (block->balance ().number (), node.ledger.confirmed.block_balance (transaction, head).value_or (0).number ());
-	auto const time_priority = !head.is_zero () ? node.ledger.confirmed.block_get (transaction, head)->sideband ().timestamp : nano::seconds_since_epoch (); // New accounts get current timestamp i.e. lowest priority
+	return activate (transaction, block);
+}
+
+std::shared_ptr<nano::block> nano::scheduler::priority::activate (secure::transaction const & transaction, std::shared_ptr<nano::block> const & block)
+{
+	auto account = block->account ();
+	auto head = node.ledger.confirmed.account_head (transaction, account);
+	auto const balance_priority = std::max (block->balance (), node.ledger.any.block_balance (transaction, block->previous ()).value_or (0));
+	auto timestamp_calculation = [&] () {
+		std::chrono::milliseconds diff{ 0 };
+		if (!head.is_zero ())
+		{
+			auto timestamp = node.ledger.confirmed.block_get (transaction, head)->sideband ().timestamp;
+			diff = std::chrono::seconds{ nano::seconds_since_epoch () - timestamp };
+		}
+		// Use clock with higher precision than seconds
+		auto time = std::chrono::steady_clock::now () - diff; // New accounts get current timestamp i.e. lowest priority
+		return time;
+	};
+	auto const time_priority = timestamp_calculation ();
 
 	node.stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::activated);
 	node.logger.trace (nano::log::type::election_scheduler, nano::log::detail::block_activated,
 	nano::log::arg{ "account", account.to_account () }, // TODO: Convert to lazy eval
 	nano::log::arg{ "block", block },
-	nano::log::arg{ "time", time_priority },
+	nano::log::arg{ "time", time_priority.time_since_epoch ().count () },
 	nano::log::arg{ "priority", balance_priority });
 
-	nano::lock_guard<nano::mutex> lock{ mutex };
-	buckets->push (time_priority, block, balance_priority);
-	notify ();
-
-	return true; // Activated
-}
-
-void nano::scheduler::priority::notify ()
-{
-	condition.notify_all ();
-}
-
-std::size_t nano::scheduler::priority::size () const
-{
-	nano::lock_guard<nano::mutex> lock{ mutex };
-	return buckets->size ();
-}
+	stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::insert_priority);
+	auto result = node.active.insert (block);
+	if (result.inserted)
+	{
+		stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::insert_priority_success);
+	}
+	if (result.election != nullptr)
+	{
+		result.election->transition_active ();
+	}
 
-bool nano::scheduler::priority::empty_locked () const
-{
-	return buckets->empty ();
+	nano::unique_lock<nano::mutex> lock{ mutex };
+	if (tracking.find (block->hash ()) != tracking.end ())
+	{
+		return nullptr;
+	}
+	auto iter = buckets.upper_bound (balance_priority);
+	--iter; // Iterator points to bucket after the target priority
+	debug_assert (iter != buckets.end ());
+	auto & bucket = *iter->second;
+	auto removed = bucket.insert (time_priority, block);
+	if (removed == nullptr)
+	{
+		node.stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::block_insert);
+		node.logger.trace (nano::log::type::election_scheduler, nano::log::detail::block_insert,
+		nano::log::arg{ "block", block->hash ().to_string () },
+		nano::log::arg{ "time", time_priority.time_since_epoch ().count () });
+		// Bucket was not at full capacity
+		auto inserted = tracking.emplace (block->hash (), &bucket);
+		lock.unlock ();
+		debug_assert (inserted.second);
+		return nullptr;
+	}
+	else if (removed != block)
+	{
+		node.stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::block_overflow);
+		node.logger.trace (nano::log::type::election_scheduler, nano::log::detail::block_overflow,
+		nano::log::arg{ "block", block->hash ().to_string () },
+		nano::log::arg{ "removed", removed->hash ().to_string () },
+		nano::log::arg{ "time", time_priority.time_since_epoch ().count () });
+		// Bucket was full and another block was lowest priority
+		auto inserted = tracking.emplace (block->hash (), &bucket);
+		lock.unlock ();
+		node.active.erase (*removed);
+		debug_assert (inserted.second);
+		return removed;
+	}
+	else
+	{
+		node.stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::block_reject);
+		node.logger.trace (nano::log::type::election_scheduler, nano::log::detail::block_reject,
+		nano::log::arg{ "block", block->hash ().to_string () },
+		nano::log::arg{ "removed", removed->hash ().to_string () },
+		nano::log::arg{ "time", time_priority.time_since_epoch ().count () });
+		lock.unlock ();
+		// Bucket was full and block inserted was lowest priority
+		node.active.erase (*block);
+		return block;
+	}
 }
 
-bool nano::scheduler::priority::empty () const
+void nano::scheduler::priority::election_stopped (std::shared_ptr<nano::election> election)
 {
 	nano::lock_guard<nano::mutex> lock{ mutex };
-	return empty_locked ();
-}
-
-bool nano::scheduler::priority::predicate () const
-{
-	return node.active.vacancy () > 0 && !buckets->empty ();
+	for (auto const & [hash, block] : election->blocks ())
+	{
+		if (auto existing = tracking.find (hash); existing != tracking.end ())
+		{
+			auto erased = existing->second->erase (hash);
+			debug_assert (erased == 1);
+			tracking.erase (existing);
+		}
+	}
 }
 
-void nano::scheduler::priority::run ()
+void nano::scheduler::priority::setup_buckets ()
 {
-	nano::unique_lock<nano::mutex> lock{ mutex };
-	while (!stopped)
-	{
-		condition.wait (lock, [this] () {
-			return stopped || predicate ();
-		});
-		debug_assert ((std::this_thread::yield (), true)); // Introduce some random delay in debug builds
-		if (!stopped)
+	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)
 		{
-			stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::loop);
-
-			if (predicate ())
-			{
-				auto block = buckets->top ();
-				buckets->pop ();
-				lock.unlock ();
-				stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::insert_priority);
-				auto result = node.active.insert (block);
-				if (result.inserted)
-				{
-					stats.inc (nano::stat::type::election_scheduler, nano::stat::detail::insert_priority_success);
-				}
-				if (result.election != nullptr)
-				{
-					result.election->transition_active ();
-				}
-			}
-			else
-			{
-				lock.unlock ();
-			}
-			notify ();
-			lock.lock ();
+			buckets.emplace (begin + i * width, std::make_unique<nano::scheduler::bucket> (node.config.priority_scheduler.depth));
 		}
-	}
+	};
+	build_region (0, 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);
+	build_region (uint128_t{ 1 } << 120, uint128_t{ 1 } << 127, 1);
 }
 
 std::unique_ptr<nano::container_info_component> nano::scheduler::priority::collect_container_info (std::string const & name)
@@ -142,6 +161,5 @@ std::unique_ptr<nano::container_info_component> nano::scheduler::priority::colle
 	nano::unique_lock<nano::mutex> lock{ mutex };
 
 	auto composite = std::make_unique<container_info_composite> (name);
-	composite->add_component (buckets->collect_container_info ("buckets"));
 	return composite;
 }
diff --git a/nano/node/scheduler/priority.hpp b/nano/node/scheduler/priority.hpp
index 8de27eaae6..20fcbb1cbb 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>
@@ -14,6 +15,7 @@ namespace nano
 {
 class block;
 class container_info_component;
+class election;
 class node;
 class stats;
 }
@@ -21,27 +23,42 @@ namespace nano::secure
 {
 class transaction;
 }
+namespace nano::scheduler
+{
+class bucket;
+}
 
 namespace nano::scheduler
 {
-class buckets;
+class priority_config final
+{
+public:
+	// nano::error deserialize (nano::tomlconfig & toml);
+	// nano::error serialize (nano::tomlconfig & toml) const;
+
+public:
+	size_t depth{ 4096 };
+};
 class priority final
 {
 public:
 	priority (nano::node &, nano::stats &);
 	~priority ();
 
-	void start ();
-	void stop ();
-
 	/**
-	 * Activates the first unconfirmed block of \p account_a
-	 * @return true if account was activated
+	 * Activates the first unconfirmed block of \p account
+	 * @return Block that was evicted when the first unconfirmed block for \p account was activated
+	 */
+	std::shared_ptr<nano::block> activate (secure::transaction const & transaction, nano::account const & account);
+	/**
+	 * Activates the block \p block
+	 * @return Block that was evicted when \p block was activated
 	 */
-	bool activate (secure::transaction const &, nano::account const &);
-	void notify ();
-	std::size_t size () const;
-	bool empty () const;
+	std::shared_ptr<nano::block> activate (secure::transaction const & transaction, std::shared_ptr<nano::block> const & block);
+	/**
+	 * Notify container when election has stopped to free space
+	 */
+	void election_stopped (std::shared_ptr<nano::election> election);
 
 	std::unique_ptr<container_info_component> collect_container_info (std::string const & name);
 
@@ -50,15 +67,10 @@ class priority final
 	nano::stats & stats;
 
 private:
-	void run ();
-	bool empty_locked () const;
-	bool predicate () const;
-
-	std::unique_ptr<nano::scheduler::buckets> buckets;
+	void setup_buckets ();
 
-	bool stopped{ false };
-	nano::condition_variable condition;
+	std::unordered_map<nano::block_hash, nano::scheduler::bucket *> tracking;
+	std::map<nano::amount, std::unique_ptr<nano::scheduler::bucket>> buckets;
 	mutable nano::mutex mutex;
-	std::thread thread;
 };
 }
diff --git a/nano/rpc_test/rpc.cpp b/nano/rpc_test/rpc.cpp
index 37e3e66cf2..3093ca9ab6 100644
--- a/nano/rpc_test/rpc.cpp
+++ b/nano/rpc_test/rpc.cpp
@@ -6962,7 +6962,7 @@ TEST (rpc, election_statistics)
 	request.put ("action", "election_statistics");
 
 	auto response = wait_response (system, rpc_ctx, request);
-	ASSERT_EQ ("1", response.get<std::string> ("normal"));
+	ASSERT_EQ ("1", response.get<std::string> ("priority"));
 	ASSERT_EQ ("0", response.get<std::string> ("hinted"));
 	ASSERT_EQ ("0", response.get<std::string> ("optimistic"));
 	ASSERT_EQ ("1", response.get<std::string> ("total"));
diff --git a/nano/secure/common.hpp b/nano/secure/common.hpp
index fd28d30b13..a2b912faa3 100644
--- a/nano/secure/common.hpp
+++ b/nano/secure/common.hpp
@@ -206,7 +206,8 @@ enum class block_status
 	balance_mismatch, // Balance and amount delta don't match
 	representative_mismatch, // Representative is changed when it is not allowed
 	block_position, // This block cannot follow the previous block
-	insufficient_work // Insufficient work for this block, even though it passed the minimal validation
+	insufficient_work, // Insufficient work for this block, even though it passed the minimal validation
+	overflow // Insufficient space in scheduler
 };
 
 std::string_view to_string (block_status);
diff --git a/nano/slow_test/node.cpp b/nano/slow_test/node.cpp
index f5e17a6a4f..27911f9d33 100644
--- a/nano/slow_test/node.cpp
+++ b/nano/slow_test/node.cpp
@@ -2166,7 +2166,7 @@ TEST (system, block_sequence)
 			std::string message;
 			for (auto i : system.nodes)
 			{
-				message += boost::str (boost::format ("N:%1% b:%2% c:%3% a:%4% s:%5% p:%6%\n") % std::to_string (i->network.port) % std::to_string (i->ledger.block_count ()) % std::to_string (i->ledger.cemented_count ()) % std::to_string (i->active.size ()) % std::to_string (i->scheduler.priority.size ()) % std::to_string (i->network.size ()));
+				message += boost::str (boost::format ("N:%1% b:%2% c:%3% a:%4% p:%5%\n") % std::to_string (i->network.port) % std::to_string (i->ledger.block_count ()) % std::to_string (i->ledger.cemented_count ()) % std::to_string (i->active.size ()) % std::to_string (i->network.size ()));
 				nano::lock_guard<nano::mutex> lock{ i->active.mutex };
 				for (auto const & j : i->active.roots)
 				{