diff --git a/nano/core_test/node.cpp b/nano/core_test/node.cpp
index 65ba12a705..241126f988 100644
--- a/nano/core_test/node.cpp
+++ b/nano/core_test/node.cpp
@@ -3076,6 +3076,8 @@ TEST (node, block_processor_half_full)
 	node_flags.force_use_write_database_queue = true;
 	auto & node = *system.add_node (nano::node_config (system.get_available_port (), system.logging), node_flags);
 	nano::state_block_builder builder;
+
+	// All blocks are forks of each other so none of the blocks going through pipeline have a gap
 	auto send1 = builder.make_block ()
 				 .account (nano::dev::genesis_key.pub)
 				 .previous (nano::dev::genesis->hash ())
@@ -3087,24 +3089,23 @@ TEST (node, block_processor_half_full)
 				 .build_shared ();
 	auto send2 = builder.make_block ()
 				 .account (nano::dev::genesis_key.pub)
-				 .previous (send1->hash ())
+				 .previous (nano::dev::genesis->hash ())
 				 .representative (nano::dev::genesis_key.pub)
 				 .balance (nano::dev::constants.genesis_amount - 2 * nano::Gxrb_ratio)
 				 .link (nano::dev::genesis_key.pub)
 				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
-				 .work (*node.work_generate_blocking (send1->hash ()))
+				 .work (*node.work_generate_blocking (nano::dev::genesis->hash ()))
 				 .build_shared ();
 	auto send3 = builder.make_block ()
 				 .account (nano::dev::genesis_key.pub)
-				 .previous (send2->hash ())
+				 .previous (nano::dev::genesis->hash ())
 				 .representative (nano::dev::genesis_key.pub)
 				 .balance (nano::dev::constants.genesis_amount - 3 * nano::Gxrb_ratio)
 				 .link (nano::dev::genesis_key.pub)
 				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
-				 .work (*node.work_generate_blocking (send2->hash ()))
+				 .work (*node.work_generate_blocking (nano::dev::genesis->hash ()))
 				 .build_shared ();
-	// The write guard prevents block processor doing any writes
-	auto write_guard = node.write_database_queue.wait (nano::writer::testing);
+	node.block_processor.stop ();
 	node.block_processor.add (send1);
 	ASSERT_FALSE (node.block_processor.half_full ());
 	node.block_processor.add (send2);
diff --git a/nano/node/CMakeLists.txt b/nano/node/CMakeLists.txt
index 3846c78973..4dfe90ecc3 100644
--- a/nano/node/CMakeLists.txt
+++ b/nano/node/CMakeLists.txt
@@ -24,6 +24,10 @@ add_library(
   block_arrival.cpp
   block_broadcast.cpp
   block_broadcast.hpp
+  block_checker.cpp
+  block_checker.hpp
+  block_checker_rules.cpp
+  block_checker_rules.hpp
   block_publisher.cpp
   block_publisher.hpp
   gap_tracker.cpp
diff --git a/nano/node/block_checker.cpp b/nano/node/block_checker.cpp
new file mode 100644
index 0000000000..ca9ac0b467
--- /dev/null
+++ b/nano/node/block_checker.cpp
@@ -0,0 +1,225 @@
+#include <nano/node/block_checker.hpp>
+#include <nano/node/block_checker_rules.hpp>
+#include <nano/secure/ledger.hpp>
+#include <nano/store/block.hpp>
+#include <nano/store/component.hpp>
+#include <nano/store/pending.hpp>
+
+nano::block_checker::context::context (nano::ledger & ledger, std::shared_ptr<nano::block> block) :
+	block{ block },
+	epochs{ ledger.constants.epochs }
+{
+	auto transaction = ledger.store.tx_begin_read ();
+	previous = ledger.store.block.get (transaction, block->previous ());
+	if (!gap_previous ())
+	{
+		state = ledger.account_info (transaction, account ());
+		if (!state)
+		{
+			state = nano::account_info{};
+		}
+		source_exists = ledger.block_or_pruned_exists (transaction, source ());
+		pending = ledger.pending_info (transaction, { account (), source () });
+		any_pending = ledger.store.pending.any (transaction, account ());
+	}
+	if (ledger.store.block.exists (transaction, block->hash ()))
+	{
+		this->block = nullptr; // Signal this block already exists by nulling out block
+	}
+	//nano::account const & account_a, nano::block_hash const & successor_a, nano::amount const & balance_a, uint64_t const height_a, nano::seconds_t const timestamp_a, nano::block_details const & details_a, nano::epoch const source_epoch_a
+	//block->sideband_set (nano::block_sideband (account (), 0, pending.amount, 1, nano::seconds_since_epoch (), block_details, nano::epoch::epoch_0 /* unused */));
+}
+/**
+ 	  This class filters blocks in four directions based on how the link field should be interpreted
+ 	  For state blocks the link field is interpreted as:
+ 	    If the balance has decreased, a destination account
+ 	    If the balance has not decreased
+ 	      If the link field is 0, a noop
+ 	      If the link field is an epoch link, an epoch sentinel
+ 	      Otherwise, a block hash of an block ready to be received
+ 	  For legacy blocks, the link field interpretation is applied to source field for receive and open blocks or the destination field for send blocks */
+nano::block_checker::block_op nano::block_checker::context::block_op () const
+{
+	debug_assert (state.has_value ());
+	switch (block->type ())
+	{
+		case nano::block_type::state:
+			if (block->balance () < state->balance)
+			{
+				return nano::block_checker::block_op::send;
+			}
+			if (block->link ().is_zero ())
+			{
+				return nano::block_checker::block_op::noop;
+			}
+			if (epochs.is_epoch_link (block->link ()))
+			{
+				return nano::block_checker::block_op::epoch;
+			}
+			return nano::block_checker::block_op::receive;
+		case nano::block_type::send:
+			return nano::block_checker::block_op::send;
+		case nano::block_type::open:
+		case nano::block_type::receive:
+			return nano::block_checker::block_op::receive;
+		case nano::block_type::change:
+			return nano::block_checker::block_op::noop;
+		case nano::block_type::not_a_block:
+		case nano::block_type::invalid:
+			release_assert (false);
+			break;
+	}
+	release_assert (false);
+}
+
+bool nano::block_checker::context::is_send () const
+{
+	debug_assert (state.has_value ());
+	auto legacy_send = block->type () == nano::block_type::send;
+	auto type = block->type () == nano::block_type::state;
+	auto decreased = block->balance () < state->balance;
+	return legacy_send || (type && decreased);
+}
+
+nano::account nano::block_checker::context::account () const
+{
+	switch (block->type ())
+	{
+		case nano::block_type::change:
+		case nano::block_type::receive:
+		case nano::block_type::send:
+			debug_assert (previous != nullptr);
+			switch (previous->type ())
+			{
+				case nano::block_type::state:
+				case nano::block_type::open:
+					return previous->account ();
+				case nano::block_type::change:
+				case nano::block_type::receive:
+				case nano::block_type::send:
+					return previous->sideband ().account;
+				case nano::block_type::not_a_block:
+				case nano::block_type::invalid:
+					debug_assert (false);
+					break;
+			}
+			break;
+		case nano::block_type::state:
+		case nano::block_type::open:
+			return block->account ();
+		case nano::block_type::not_a_block:
+		case nano::block_type::invalid:
+			debug_assert (false);
+			break;
+	}
+	// std::unreachable (); c++23
+	return 1; // Return an account that cannot be signed for.
+}
+
+nano::block_hash nano::block_checker::context::source () const
+{
+	switch (block->type ())
+	{
+		case nano::block_type::send:
+		case nano::block_type::change:
+			// 0 is returned for source on send/change blocks
+		case nano::block_type::receive:
+		case nano::block_type::open:
+			return block->source ();
+		case nano::block_type::state:
+			return block->link ().as_block_hash ();
+		case nano::block_type::not_a_block:
+		case nano::block_type::invalid:
+			return 0;
+	}
+	debug_assert (false);
+	return 0;
+}
+
+nano::account nano::block_checker::context::signer (nano::epochs const & epochs) const
+{
+	debug_assert (block != nullptr);
+	switch (block->type ())
+	{
+		case nano::block_type::send:
+		case nano::block_type::receive:
+		case nano::block_type::change:
+			debug_assert (previous != nullptr); // Previous block must be passed in for non-open blocks
+			switch (previous->type ())
+			{
+				case nano::block_type::state:
+					debug_assert (false && "Legacy blocks can't follow state blocks");
+					break;
+				case nano::block_type::open:
+					// Open blocks have the account written in the block.
+					return previous->account ();
+				default:
+					// Other legacy block types have the account stored in sideband.
+					return previous->sideband ().account;
+			}
+			break;
+		case nano::block_type::state:
+		{
+			debug_assert (dynamic_cast<nano::state_block *> (block.get ()));
+			// If the block is a send, while the link field may contain an epoch link value, it is actually a malformed destination address.
+			return (!epochs.is_epoch_link (block->link ()) || is_send ()) ? block->account () : epochs.signer (epochs.epoch (block->link ()));
+		}
+		case nano::block_type::open: // Open block signer is determined statelessly as it's written in the block
+			return block->account ();
+		case nano::block_type::invalid:
+		case nano::block_type::not_a_block:
+			debug_assert (false);
+			break;
+	}
+	// std::unreachable (); c++23
+	return 1; // Return an account that cannot be signed for.
+}
+
+bool nano::block_checker::context::gap_previous () const
+{
+	return !block->previous ().is_zero () && previous == nullptr;
+}
+
+nano::block_checker::block_checker ()
+{
+}
+
+nano::process_result nano::block_checker::check (context & context)
+{
+	if (!context.block)
+	{
+		return nano::process_result::old;
+	}
+	nano::process_result result;
+	if (result = nano::rule_reserved_account (context), result != nano::process_result::progress)
+	{
+		return result;
+	}
+	if (result = nano::rule_previous_frontier (context), result != nano::process_result::progress)
+	{
+		return result;
+	}
+	if (result = nano::rule_block_position (context), result != nano::process_result::progress)
+	{
+		return result;
+	}
+	if (result = nano::rule_block_signed (context), result != nano::process_result::progress)
+	{
+		return result;
+	}
+	if (result = nano::rule_metastable (context), result != nano::process_result::progress)
+	{
+		return result;
+	}
+	switch (context.block_op ())
+	{
+		case block_op::receive:
+			return nano::rule_receivable (context);
+		case block_op::send:
+			return nano::rule_send_restrictions (context);
+		case block_op::noop:
+			return nano::process_result::progress;
+		case block_op::epoch:
+			return nano::rule_epoch_restrictions (context);
+	}
+}
diff --git a/nano/node/block_checker.hpp b/nano/node/block_checker.hpp
new file mode 100644
index 0000000000..cdd79fdab0
--- /dev/null
+++ b/nano/node/block_checker.hpp
@@ -0,0 +1,52 @@
+#pragma once
+
+#include <nano/lib/epoch.hpp>
+#include <nano/lib/numbers.hpp>
+#include <nano/secure/common.hpp>
+
+#include <memory>
+#include <optional>
+
+namespace nano
+{
+class block;
+class ledger;
+}
+
+namespace nano
+{
+class block_checker
+{
+public:
+	enum class block_op
+	{
+		receive,
+		send,
+		noop,
+		epoch
+	};
+	// Context that is passed between pipeline stages
+	class context
+	{
+	public:
+		context (nano::ledger & ledger, std::shared_ptr<nano::block> block);
+		bool is_send () const;
+		nano::account account () const;
+		nano::block_hash source () const;
+		nano::account signer (nano::epochs const & epochs) const;
+		bool gap_previous () const;
+		block_op block_op () const;
+		std::shared_ptr<nano::block> block;
+		std::shared_ptr<nano::block> previous;
+		std::optional<nano::account_info> state;
+		std::optional<nano::pending_info> pending;
+		bool any_pending{ false };
+		bool source_exists{ false };
+		nano::epochs & epochs;
+	};
+
+public:
+	block_checker ();
+	nano::process_result check (context & context);
+};
+} // namespace nano
diff --git a/nano/node/block_checker_rules.cpp b/nano/node/block_checker_rules.cpp
new file mode 100644
index 0000000000..284caa68bb
--- /dev/null
+++ b/nano/node/block_checker_rules.cpp
@@ -0,0 +1,142 @@
+#include <nano/node/block_checker_rules.hpp>
+
+nano::process_result nano::rule_reserved_account (nano::block_checker::context & context)
+{
+	switch (context.block->type ())
+	{
+		case nano::block_type::open:
+		case nano::block_type::state:
+			if (!context.block->account ().is_zero ())
+			{
+				return nano::process_result::progress;
+			}
+			else
+			{
+				return nano::process_result::opened_burn_account;
+			}
+			break;
+		case nano::block_type::change:
+		case nano::block_type::receive:
+		case nano::block_type::send:
+			return nano::process_result::progress;
+		case nano::block_type::invalid:
+		case nano::block_type::not_a_block:
+			release_assert (false);
+			break;
+	}
+	release_assert (false);
+}
+
+nano::process_result nano::rule_previous_frontier (nano::block_checker::context & context)
+{
+	if (context.block == nullptr)
+	{
+		return nano::process_result::old;
+	}
+	else if (context.gap_previous ())
+	{
+		return nano::process_result::gap_previous;
+	}
+	else
+	{
+		return nano::process_result::progress;
+	}
+}
+
+nano::process_result nano::rule_block_position (nano::block_checker::context & context)
+{
+	auto const & block = context.block;
+	auto const & previous = context.previous;
+	if (previous == nullptr)
+	{
+		return nano::process_result::progress;
+	}
+	switch (block->type ())
+	{
+		case nano::block_type::send:
+		case nano::block_type::receive:
+		case nano::block_type::change:
+		{
+			switch (previous->type ())
+			{
+				case nano::block_type::state:
+					return nano::process_result::block_position;
+				default:
+					return nano::process_result::progress;
+			}
+		}
+		default:
+			return nano::process_result::progress;
+	}
+}
+
+nano::process_result nano::rule_block_signed (nano::block_checker::context & context)
+{
+	if (!nano::validate_message (context.signer (context.epochs), context.block->hash (), context.block->block_signature ()))
+	{
+		return nano::process_result::progress;
+	}
+	return nano::process_result::bad_signature;
+}
+
+nano::process_result nano::rule_metastable (nano::block_checker::context & context)
+{
+	debug_assert (context.state.has_value ());
+	if (context.block->previous () == context.state->head)
+	{
+		return nano::process_result::progress;
+	}
+	else
+	{
+		return nano::process_result::fork;
+	}
+}
+
+nano::process_result nano::rule_receivable (nano::block_checker::context & context)
+{
+	if (!context.source_exists)
+	{
+		return nano::process_result::gap_source;
+	}
+	if (!context.pending.has_value ())
+	{
+		return nano::process_result::unreceivable;
+	}
+	if (context.block->type () == nano::block_type::state)
+	{
+		auto next_balance = context.state->balance.number () + context.pending->amount.number ();
+		if (next_balance != context.block->balance ().number ())
+		{
+			return nano::process_result::balance_mismatch;
+		}
+	}
+	return nano::process_result::progress;
+}
+
+nano::process_result nano::rule_epoch_restrictions (nano::block_checker::context & context)
+{
+	debug_assert (context.state.has_value ());
+	if (context.state->balance != context.block->balance ())
+	{
+		return nano::process_result::balance_mismatch;
+	}
+	if (context.state->representative != context.block->representative ())
+	{
+		return nano::process_result::representative_mismatch;
+	}
+	if (context.block->previous ().is_zero () && !context.any_pending)
+	{
+		return nano::process_result::gap_epoch_open_pending;
+	}
+	return nano::process_result::progress;
+}
+
+nano::process_result nano::rule_send_restrictions (nano::block_checker::context & context)
+{
+	debug_assert (context.block->type () == nano::block_type::send || context.block->type () == nano::block_type::state);
+	if (context.state->balance < context.block->balance ())
+	{
+		return nano::process_result::negative_spend;
+	}
+	return nano::process_result::progress;
+}
diff --git a/nano/node/block_checker_rules.hpp b/nano/node/block_checker_rules.hpp
new file mode 100644
index 0000000000..98d8d08ca5
--- /dev/null
+++ b/nano/node/block_checker_rules.hpp
@@ -0,0 +1,37 @@
+#pragma once
+
+#include <nano/node/block_checker.hpp>
+
+namespace nano
+{
+class ledger;
+}
+
+namespace nano
+{
+/**
+ 	Filters accounts that are reserved and cannot be used e.g. the account number 0.
+  */
+nano::process_result rule_reserved_account (nano::block_checker::context & context);
+/**
+ 	  This rule checks if the previous block for this block is the head block of the specified account
+  */
+nano::process_result rule_previous_frontier (nano::block_checker::context & context);
+
+nano::process_result rule_block_position (nano::block_checker::context & context);
+nano::process_result rule_block_signed (nano::block_checker::context & context);
+
+/**
+ 	  This rule identifies metastable blocks (forked blocks) with respect to the ledger and rejects them.
+ 	  Rejected blocks need to be resolved via consensus
+ 	  It is assumed that the previous block has already been loaded in to `context' if it exists
+ 	  Metastable scenarios are:
+ 	    1) An initial block arriving for an account that's already been initialized
+ 	    2) The previous block exists but it is not the head block
+ 	  Both of these scenarios can be ifentified by checking: if block->previous () == head
+ 	 */
+nano::process_result rule_metastable (nano::block_checker::context & context);
+nano::process_result rule_receivable (nano::block_checker::context & context);
+nano::process_result rule_epoch_restrictions (nano::block_checker::context & context);
+nano::process_result rule_send_restrictions (nano::block_checker::context & context);
+}
diff --git a/nano/node/blockprocessor.cpp b/nano/node/blockprocessor.cpp
index 6d843fea54..034b9504a3 100644
--- a/nano/node/blockprocessor.cpp
+++ b/nano/node/blockprocessor.cpp
@@ -1,5 +1,6 @@
 #include <nano/lib/threading.hpp>
 #include <nano/lib/timer.hpp>
+#include <nano/node/block_checker.hpp>
 #include <nano/node/blockprocessor.hpp>
 #include <nano/node/node.hpp>
 #include <nano/store/component.hpp>
@@ -158,6 +159,7 @@ void nano::block_processor::process_blocks ()
 			active = true;
 			lock.unlock ();
 			auto processed = process_batch (lock);
+			queue_unchecked (processed);
 			batch_processed.notify (processed);
 			lock.lock ();
 			active = false;
@@ -196,11 +198,40 @@ bool nano::block_processor::have_blocks ()
 
 void nano::block_processor::add_impl (std::shared_ptr<nano::block> block)
 {
+	nano::block_checker::context context{ node.ledger, block };
+	nano::block_checker checker;
+	auto code = checker.check (context);
+	switch (code)
 	{
-		nano::lock_guard<nano::mutex> guard{ mutex };
-		blocks.emplace_back (block);
+		case nano::process_result::progress:
+		{
+			nano::lock_guard<nano::mutex> guard{ mutex };
+			blocks.emplace_back (block);
+			condition.notify_all ();
+			return;
+		}
+		case nano::process_result::gap_previous:
+			node.gap_cache.add (context.block->hash ());
+			node.unchecked.put (context.block->previous (), context.block);
+			break;
+		case nano::process_result::gap_source:
+			node.gap_cache.add (context.block->hash ());
+			node.unchecked.put (context.source (), context.block);
+			break;
+		case nano::process_result::gap_epoch_open_pending:
+			node.unchecked.put (context.account (), context.block);
+			break;
+		case nano::process_result::fork:
+			node.active.publish (context.block);
+			break;
+		case nano::process_result::old:
+			// drop
+			break;
+		default:
+			// drop
+			break;
 	}
-	condition.notify_all ();
+	processed.notify (nano::process_return{ code }, block);
 }
 
 auto nano::block_processor::process_batch (nano::unique_lock<nano::mutex> & lock_a) -> std::deque<processed_t>
@@ -273,16 +304,6 @@ nano::process_return nano::block_processor::process_one (store::write_transactio
 				block->serialize_json (block_string, node.config.logging.single_line_record ());
 				node.logger.try_log (boost::str (boost::format ("Processing block %1%: %2%") % hash.to_string () % block_string));
 			}
-			queue_unchecked (transaction_a, hash);
-			/* For send blocks check epoch open unchecked (gap pending).
-			For state blocks check only send subtype and only if block epoch is not last epoch.
-			If epoch is last, then pending entry shouldn't trigger same epoch open block for destination account. */
-			if (block->type () == nano::block_type::send || (block->type () == nano::block_type::state && block->sideband ().details.is_send && std::underlying_type_t<nano::epoch> (block->sideband ().details.epoch) < std::underlying_type_t<nano::epoch> (nano::epoch::max)))
-			{
-				/* block->destination () for legacy send blocks
-				block->link () for state blocks (send subtype) */
-				queue_unchecked (transaction_a, block->destination ().is_zero () ? block->link () : block->destination ());
-			}
 			break;
 		}
 		case nano::process_result::gap_previous:
@@ -404,10 +425,29 @@ nano::process_return nano::block_processor::process_one (store::write_transactio
 	return result;
 }
 
-void nano::block_processor::queue_unchecked (store::write_transaction const & transaction_a, nano::hash_or_account const & hash_or_account_a)
+void nano::block_processor::queue_unchecked (std::deque<processed_t> & processed)
 {
-	node.unchecked.trigger (hash_or_account_a);
-	node.gap_cache.erase (hash_or_account_a.hash);
+	auto queue_operation = [this] (nano::hash_or_account const & hash_or_account) {
+		node.unchecked.trigger (hash_or_account);
+		node.gap_cache.erase (hash_or_account.hash);
+	};
+	for (auto i : processed)
+	{
+		if (i.first.code == nano::process_result::progress)
+		{
+			auto block = i.second;
+			queue_operation (block->hash ());
+			/* For send blocks check epoch open unchecked (gap pending).
+			For state blocks check only send subtype and only if block epoch is not last epoch.
+			If epoch is last, then pending entry shouldn't trigger same epoch open block for destination account. */
+			if (block->type () == nano::block_type::send || (block->type () == nano::block_type::state && block->sideband ().details.is_send && std::underlying_type_t<nano::epoch> (block->sideband ().details.epoch) < std::underlying_type_t<nano::epoch> (nano::epoch::max)))
+			{
+				/* block->destination () for legacy send blocks
+				block->link () for state blocks (send subtype) */
+				queue_operation (block->destination ().is_zero () ? block->link () : block->destination ());
+			}
+		}
+	}
 }
 
 std::unique_ptr<nano::container_info_component> nano::collect_container_info (block_processor & block_processor, std::string const & name)
diff --git a/nano/node/blockprocessor.hpp b/nano/node/blockprocessor.hpp
index 5e87663897..c575f2abdb 100644
--- a/nano/node/blockprocessor.hpp
+++ b/nano/node/blockprocessor.hpp
@@ -56,7 +56,7 @@ class block_processor final
 	// Roll back block in the ledger that conflicts with 'block'
 	void rollback_competitor (store::write_transaction const & transaction, nano::block const & block);
 	nano::process_return process_one (store::write_transaction const &, std::shared_ptr<nano::block> block, bool const = false);
-	void queue_unchecked (store::write_transaction const &, nano::hash_or_account const &);
+	void queue_unchecked (std::deque<processed_t> & processed);
 	std::deque<processed_t> process_batch (nano::unique_lock<nano::mutex> &);
 	void add_impl (std::shared_ptr<nano::block> block);
 	bool stopped{ false };