diff --git a/benchmarks/block_ref.py b/benchmarks/block_ref.py
index beb5a900d8f0..ca65202ff157 100644
--- a/benchmarks/block_ref.py
+++ b/benchmarks/block_ref.py
@@ -19,6 +19,7 @@
 from chia.consensus.get_block_generator import get_block_generator
 from chia.full_node.block_store import BlockStore
 from chia.full_node.coin_store import CoinStore
+from chia.full_node.consensus_store_sqlite3 import ConsensusStoreSQLite3
 from chia.types.blockchain_format.serialized_program import SerializedProgram
 from chia.util.db_version import lookup_db_version
 from chia.util.db_wrapper import DBWrapper2
@@ -70,7 +71,8 @@ async def main(db_path: Path) -> None:
         # make configurable
         reserved_cores = 4
         height_map = await BlockHeightMap.create(db_path.parent, db_wrapper)
-        blockchain = await Blockchain.create(coin_store, block_store, height_map, DEFAULT_CONSTANTS, reserved_cores)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        blockchain = await Blockchain.create(consensus_store, DEFAULT_CONSTANTS, reserved_cores)
 
         peak = blockchain.get_peak()
         assert peak is not None
diff --git a/chia/_tests/blockchain/blockchain_test_utils.py b/chia/_tests/blockchain/blockchain_test_utils.py
index 4bc137e5c258..d03722b90eee 100644
--- a/chia/_tests/blockchain/blockchain_test_utils.py
+++ b/chia/_tests/blockchain/blockchain_test_utils.py
@@ -15,31 +15,20 @@
 
 
 async def check_block_store_invariant(bc: Blockchain):
-    db_wrapper = bc.block_store.db_wrapper
-
-    if db_wrapper.db_version == 1:
-        return
-
     in_chain = set()
     max_height = -1
-    async with bc.block_store.transaction() as conn:
-        async with conn.execute("SELECT height, in_main_chain FROM full_blocks") as cursor:
-            rows = await cursor.fetchall()
-            for row in rows:
-                height = row[0]
-
-                # if this block is in-chain, ensure we haven't found another block
-                # at this height that's also in chain. That would be an invariant
-                # violation
-                if row[1]:
-                    # make sure we don't have any duplicate heights. Each block
-                    # height can only have a single block with in_main_chain set
-                    assert height not in in_chain
-                    in_chain.add(height)
-                    max_height = max(max_height, height)
-
-            # make sure every height is represented in the set
-            assert len(in_chain) == max_height + 1
+    async for height in bc.consensus_store.get_block_heights_in_main_chain():
+        # if this block is in-chain, ensure we haven't found another block
+        # at this height that's also in chain. That would be an invariant
+        # violation
+        # make sure we don't have any duplicate heights. Each block
+        # height can only have a single block with in_main_chain set
+        assert height not in in_chain
+        in_chain.add(height)
+        max_height = max(max_height, height)
+
+    # make sure every height is represented in the set
+    assert len(in_chain) == max_height + 1
 
 
 async def _validate_and_add_block(
diff --git a/chia/_tests/blockchain/test_blockchain.py b/chia/_tests/blockchain/test_blockchain.py
index c270d771d48b..eed56356da85 100644
--- a/chia/_tests/blockchain/test_blockchain.py
+++ b/chia/_tests/blockchain/test_blockchain.py
@@ -2076,7 +2076,8 @@ async def test_timelock_conditions(
 
             if expected == AddBlockResult.NEW_PEAK:
                 # ensure coin was in fact spent
-                c = await b.coin_store.get_coin_record(coin.name())
+                recs = await b.consensus_store.get_coin_records([coin.name()])
+                c = recs[0] if len(recs) > 0 else None
                 assert c is not None and c.spent
 
     @pytest.mark.anyio
@@ -2286,10 +2287,12 @@ async def test_ephemeral_timelock(
 
             if expected == AddBlockResult.NEW_PEAK:
                 # ensure coin1 was in fact spent
-                c = await b.coin_store.get_coin_record(coin1.name())
+                recs1 = await b.consensus_store.get_coin_records([coin1.name()])
+                c = recs1[0] if len(recs1) > 0 else None
                 assert c is not None and c.spent
                 # ensure coin2 was NOT spent
-                c = await b.coin_store.get_coin_record(coin2.name())
+                recs2 = await b.consensus_store.get_coin_records([coin2.name()])
+                c = recs2[0] if len(recs2) > 0 else None
                 assert c is not None and not c.spent
 
     @pytest.mark.anyio
@@ -3103,9 +3106,11 @@ async def test_double_spent_in_reorg(self, empty_blockchain: Blockchain, bt: Blo
             )
 
         # ephemeral coin is spent
-        first_coin = await b.coin_store.get_coin_record(new_coin.name())
+        recs_first = await b.consensus_store.get_coin_records([new_coin.name()])
+        first_coin = recs_first[0] if len(recs_first) > 0 else None
         assert first_coin is not None and first_coin.spent
-        second_coin = await b.coin_store.get_coin_record(tx_2.additions()[0].name())
+        recs_second = await b.consensus_store.get_coin_records([tx_2.additions()[0].name()])
+        second_coin = recs_second[0] if len(recs_second) > 0 else None
         assert second_coin is not None and not second_coin.spent
 
         farmer_coin = create_farmer_coin(
@@ -3121,7 +3126,8 @@ async def test_double_spent_in_reorg(self, empty_blockchain: Blockchain, bt: Blo
         )
         await _validate_and_add_block(b, blocks_reorg[-1])
 
-        farmer_coin_record = await b.coin_store.get_coin_record(farmer_coin.name())
+        recs_farmer = await b.consensus_store.get_coin_records([farmer_coin.name()])
+        farmer_coin_record = recs_farmer[0] if len(recs_farmer) > 0 else None
         assert farmer_coin_record is not None and farmer_coin_record.spent
 
     @pytest.mark.anyio
@@ -3876,11 +3882,14 @@ async def test_chain_failed_rollback(empty_blockchain: Blockchain, bt: BlockTool
         await _validate_and_add_block(b, block, expected_result=AddBlockResult.ADDED_AS_ORPHAN, fork_info=fork_info)
 
     # Incorrectly set the height as spent in DB to trigger an error
-    print(f"{await b.coin_store.get_coin_record(spend_bundle.coin_spends[0].coin.name())}")
+    recs_dbg1 = await b.consensus_store.get_coin_records([spend_bundle.coin_spends[0].coin.name()])
+    print(f"{recs_dbg1[0] if len(recs_dbg1) > 0 else None}")
     print(spend_bundle.coin_spends[0].coin.name())
-    # await b.coin_store._set_spent([spend_bundle.coin_spends[0].coin.name()], 8)
-    await b.coin_store.rollback_to_block(2)
-    print(f"{await b.coin_store.get_coin_record(spend_bundle.coin_spends[0].coin.name())}")
+    # await b.consensus_store._set_spent([spend_bundle.coin_spends[0].coin.name()], 8)
+    async with b.consensus_store as writer:
+        await writer.rollback_to_block(2)
+    recs_dbg2 = await b.consensus_store.get_coin_records([spend_bundle.coin_spends[0].coin.name()])
+    print(f"{recs_dbg2[0] if len(recs_dbg2) > 0 else None}")
 
     fork_block = blocks_reorg_chain[10 - 1]
     # fork_info = ForkInfo(fork_block.height, fork_block.height, fork_block.header_hash)
@@ -4209,7 +4218,7 @@ async def get_fork_info(blockchain: Blockchain, block: FullBlock, peak: BlockRec
     counter = 0
     start = time.monotonic()
     for height in range(fork_info.fork_height + 1, block.height):
-        fork_block: Optional[FullBlock] = await blockchain.block_store.get_full_block(fork_chain[uint32(height)])
+        fork_block: Optional[FullBlock] = await blockchain.consensus_store.get_full_block(fork_chain[uint32(height)])
         assert fork_block is not None
         assert fork_block.height - 1 == fork_info.peak_height
         assert fork_block.height == 0 or fork_block.prev_header_hash == fork_info.peak_hash
diff --git a/chia/_tests/core/full_node/ram_db.py b/chia/_tests/core/full_node/ram_db.py
index c33cc50d233c..1375f8905622 100644
--- a/chia/_tests/core/full_node/ram_db.py
+++ b/chia/_tests/core/full_node/ram_db.py
@@ -11,6 +11,7 @@
 from chia.consensus.blockchain import Blockchain
 from chia.full_node.block_store import BlockStore
 from chia.full_node.coin_store import CoinStore
+from chia.full_node.consensus_store_sqlite3 import ConsensusStoreSQLite3
 from chia.util.db_wrapper import DBWrapper2
 
 
@@ -23,7 +24,8 @@ async def create_ram_blockchain(
         block_store = await BlockStore.create(db_wrapper)
         coin_store = await CoinStore.create(db_wrapper)
         height_map = await BlockHeightMap.create(Path("."), db_wrapper)
-        blockchain = await Blockchain.create(coin_store, block_store, height_map, consensus_constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        blockchain = await Blockchain.create(consensus_store, consensus_constants, 2)
         try:
             yield db_wrapper, blockchain
         finally:
diff --git a/chia/_tests/core/full_node/stores/test_block_store.py b/chia/_tests/core/full_node/stores/test_block_store.py
index 5cb22d158f7d..e1fa53ffdc7f 100644
--- a/chia/_tests/core/full_node/stores/test_block_store.py
+++ b/chia/_tests/core/full_node/stores/test_block_store.py
@@ -24,6 +24,7 @@
 from chia.consensus.full_block_to_block_record import header_block_to_sub_block_record
 from chia.full_node.block_store import BlockStore
 from chia.full_node.coin_store import CoinStore
+from chia.full_node.consensus_store_sqlite3 import ConsensusStoreSQLite3
 from chia.full_node.full_block_utils import GeneratorBlockInfo
 from chia.simulator.block_tools import BlockTools
 from chia.simulator.wallet_tools import WalletTool
@@ -74,7 +75,8 @@ async def test_block_store(tmp_dir: Path, db_version: int, bt: BlockTools, use_c
         coin_store_2 = await CoinStore.create(db_wrapper_2)
         store_2 = await BlockStore.create(db_wrapper_2, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper_2)
-        bc = await Blockchain.create(coin_store_2, store_2, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(store_2, coin_store_2, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         await BlockStore.create(db_wrapper_2)
@@ -150,7 +152,8 @@ async def test_get_full_blocks_at(
         coin_store = await CoinStore.create(db_wrapper)
         block_store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        bc = await Blockchain.create(coin_store, block_store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         count = 0
         fork_info = ForkInfo(-1, -1, bt.constants.GENESIS_CHALLENGE)
@@ -178,7 +181,8 @@ async def test_get_block_records_in_range(
         coin_store = await CoinStore.create(db_wrapper)
         block_store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        bc = await Blockchain.create(coin_store, block_store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         count = 0
         fork_info = ForkInfo(-1, -1, bt.constants.GENESIS_CHALLENGE)
@@ -208,7 +212,8 @@ async def test_get_block_bytes_in_range_in_main_chain(
         coin_store = await CoinStore.create(db_wrapper)
         block_store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        bc = await Blockchain.create(coin_store, block_store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
         count = 0
         fork_info = ForkInfo(-1, -1, bt.constants.GENESIS_CHALLENGE)
         for b1, b2 in zip(blocks, alt_blocks):
@@ -237,7 +242,8 @@ async def test_deadlock(tmp_dir: Path, db_version: int, bt: BlockTools, use_cach
         coin_store_2 = await CoinStore.create(wrapper_2)
         store_2 = await BlockStore.create(wrapper_2)
         height_map = await BlockHeightMap.create(tmp_dir, wrapper_2)
-        bc = await Blockchain.create(coin_store_2, store_2, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(store_2, coin_store_2, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
         block_records = []
         for block in blocks:
             await _validate_and_add_block(bc, block)
@@ -268,7 +274,8 @@ async def test_rollback(bt: BlockTools, tmp_dir: Path, use_cache: bool, default_
         coin_store = await CoinStore.create(db_wrapper)
         block_store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        bc = await Blockchain.create(coin_store, block_store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         # insert all blocks
         count = 0
@@ -331,7 +338,8 @@ async def test_count_compactified_blocks(bt: BlockTools, tmp_dir: Path, db_versi
         coin_store = await CoinStore.create(db_wrapper)
         block_store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        bc = await Blockchain.create(coin_store, block_store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         count = await block_store.count_compactified_blocks()
         assert count == 0
@@ -352,7 +360,8 @@ async def test_count_uncompactified_blocks(bt: BlockTools, tmp_dir: Path, db_ver
         coin_store = await CoinStore.create(db_wrapper)
         block_store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        bc = await Blockchain.create(coin_store, block_store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         count = await block_store.count_uncompactified_blocks()
         assert count == 0
@@ -380,7 +389,8 @@ def rand_vdf_proof() -> VDFProof:
         coin_store = await CoinStore.create(db_wrapper)
         block_store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        bc = await Blockchain.create(coin_store, block_store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
         for block in blocks:
             await _validate_and_add_block(bc, block)
 
@@ -400,7 +410,7 @@ def rand_vdf_proof() -> VDFProof:
 
             # make sure we get the same result when we hit the database
             # itself (and not just the block cache)
-            block_store.rollback_cache_block(block.header_hash)
+            consensus_store.rollback_cache_block(block.header_hash)
             b = await block_store.get_full_block(block.header_hash)
             assert b is not None
             assert b.challenge_chain_ip_proof == proof
@@ -461,7 +471,8 @@ async def test_get_blocks_by_hash(tmp_dir: Path, bt: BlockTools, db_version: int
         coin_store_2 = await CoinStore.create(db_wrapper_2)
         store_2 = await BlockStore.create(db_wrapper_2, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper_2)
-        bc = await Blockchain.create(coin_store_2, store_2, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(store_2, coin_store_2, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         await BlockStore.create(db_wrapper_2)
@@ -501,7 +512,8 @@ async def test_get_block_bytes_in_range(tmp_dir: Path, bt: BlockTools, db_versio
         coin_store_2 = await CoinStore.create(db_wrapper_2)
         store_2 = await BlockStore.create(db_wrapper_2, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper_2)
-        bc = await Blockchain.create(coin_store_2, store_2, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(store_2, coin_store_2, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         await BlockStore.create(db_wrapper_2)
 
@@ -574,7 +586,8 @@ async def test_get_prev_hash(tmp_dir: Path, bt: BlockTools, db_version: int, use
         coin_store_2 = await CoinStore.create(db_wrapper_2)
         store_2 = await BlockStore.create(db_wrapper_2, use_cache=use_cache)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper_2)
-        bc = await Blockchain.create(coin_store_2, store_2, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(store_2, coin_store_2, height_map)
+        bc = await Blockchain.create(consensus_store, bt.constants, 2)
 
         store = await BlockStore.create(db_wrapper, use_cache=use_cache)
         await BlockStore.create(db_wrapper_2)
diff --git a/chia/_tests/core/full_node/stores/test_coin_store.py b/chia/_tests/core/full_node/stores/test_coin_store.py
index 1b92f1bfd73a..9d975362016e 100644
--- a/chia/_tests/core/full_node/stores/test_coin_store.py
+++ b/chia/_tests/core/full_node/stores/test_coin_store.py
@@ -22,6 +22,7 @@
 from chia.consensus.coinbase import create_farmer_coin, create_pool_coin
 from chia.full_node.block_store import BlockStore
 from chia.full_node.coin_store import CoinStore
+from chia.full_node.consensus_store_sqlite3 import ConsensusStoreSQLite3
 from chia.full_node.hint_store import HintStore
 from chia.simulator.block_tools import BlockTools, test_constants
 from chia.simulator.wallet_tools import WalletTool
@@ -318,7 +319,8 @@ async def test_basic_reorg(tmp_dir: Path, db_version: int, bt: BlockTools) -> No
         coin_store = await CoinStore.create(db_wrapper)
         store = await BlockStore.create(db_wrapper)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        b: Blockchain = await Blockchain.create(coin_store, store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(store, coin_store, height_map)
+        b: Blockchain = await Blockchain.create(consensus_store, bt.constants, 2)
         try:
             records: list[Optional[CoinRecord]] = []
 
@@ -385,7 +387,8 @@ async def test_get_puzzle_hash(tmp_dir: Path, db_version: int, bt: BlockTools) -
         coin_store = await CoinStore.create(db_wrapper)
         store = await BlockStore.create(db_wrapper)
         height_map = await BlockHeightMap.create(tmp_dir, db_wrapper)
-        b: Blockchain = await Blockchain.create(coin_store, store, height_map, bt.constants, 2)
+        consensus_store = await ConsensusStoreSQLite3.create(store, coin_store, height_map)
+        b: Blockchain = await Blockchain.create(consensus_store, bt.constants, 2)
         for block in blocks:
             await _validate_and_add_block(b, block)
         peak = b.get_peak()
diff --git a/chia/_tests/core/full_node/test_conditions.py b/chia/_tests/core/full_node/test_conditions.py
index 729b2fe67ab1..842fe477e4e6 100644
--- a/chia/_tests/core/full_node/test_conditions.py
+++ b/chia/_tests/core/full_node/test_conditions.py
@@ -83,8 +83,8 @@ async def check_spend_bundle_validity(
 
         if expected_err is None:
             await _validate_and_add_block(blockchain, newest_block)
-            coins_added = await blockchain.coin_store.get_coins_added_at_height(uint32(len(blocks)))
-            coins_removed = await blockchain.coin_store.get_coins_removed_at_height(uint32(len(blocks)))
+            coins_added = await blockchain.consensus_store.get_coins_added_at_height(uint32(len(blocks)))
+            coins_removed = await blockchain.consensus_store.get_coins_removed_at_height(uint32(len(blocks)))
         else:
             await _validate_and_add_block(blockchain, newest_block, expected_error=expected_err)
             coins_added = []
diff --git a/chia/_tests/core/full_node/test_full_node.py b/chia/_tests/core/full_node/test_full_node.py
index 89022b8f8c13..f213e56dbed6 100644
--- a/chia/_tests/core/full_node/test_full_node.py
+++ b/chia/_tests/core/full_node/test_full_node.py
@@ -47,7 +47,7 @@
 from chia.consensus.augmented_chain import AugmentedBlockchain
 from chia.consensus.block_body_validation import ForkInfo
 from chia.consensus.blockchain import Blockchain
-from chia.consensus.coin_store_protocol import CoinStoreProtocol
+from chia.consensus.consensus_store_protocol import ConsensusStoreProtocol
 from chia.consensus.get_block_challenge import get_block_challenge
 from chia.consensus.multiprocess_validation import PreValidationResult, pre_validate_block
 from chia.consensus.pot_iterations import is_overflow_block
@@ -2505,7 +2505,7 @@ def print_coin_records(records: dict[bytes32, CoinRecord]) -> None:  # pragma: n
         print(f"{rec}")
 
 
-async def validate_coin_set(coin_store: CoinStoreProtocol, blocks: list[FullBlock]) -> None:
+async def validate_coin_set(consensus_store: ConsensusStoreProtocol, blocks: list[FullBlock]) -> None:
     prev_height = blocks[0].height - 1
     prev_hash = blocks[0].prev_header_hash
     for block in blocks:
@@ -2514,7 +2514,7 @@ async def validate_coin_set(coin_store: CoinStoreProtocol, blocks: list[FullBloc
         prev_height = int(block.height)
         prev_hash = block.header_hash
         rewards = block.get_included_reward_coins()
-        records = {rec.coin.name(): rec for rec in await coin_store.get_coins_added_at_height(block.height)}
+        records = {rec.coin.name(): rec for rec in await consensus_store.get_coins_added_at_height(block.height)}
 
         # validate reward coins
         for reward in rewards:
@@ -2550,7 +2550,7 @@ async def validate_coin_set(coin_store: CoinStoreProtocol, blocks: list[FullBloc
             print_coin_records(records)
         assert records == {}
 
-        records = {rec.coin.name(): rec for rec in await coin_store.get_coins_removed_at_height(block.height)}
+        records = {rec.coin.name(): rec for rec in await consensus_store.get_coins_removed_at_height(block.height)}
         for name, rem in removals:
             rec = records.pop(name)
             assert rec is not None
@@ -2598,8 +2598,8 @@ async def test_long_reorg(
     assert reorg_blocks[fork_point] == default_10000_blocks[fork_point]
     assert reorg_blocks[fork_point + 1] != default_10000_blocks[fork_point + 1]
 
-    assert node.full_node._coin_store is not None
-    await validate_coin_set(node.full_node._coin_store, blocks)
+    assert node.full_node.blockchain.consensus_store is not None
+    await validate_coin_set(node.full_node.blockchain.consensus_store, blocks)
 
     # one aspect of this test is to make sure we can reorg blocks that are
     # not in the cache. We need to explicitly prune the cache to get that
@@ -2614,7 +2614,7 @@ async def test_long_reorg(
     chain_2_weight = peak.weight
     chain_2_peak = peak.header_hash
 
-    await validate_coin_set(node.full_node._coin_store, reorg_blocks)
+    await validate_coin_set(node.full_node.blockchain.consensus_store, reorg_blocks)
 
     # if the reorg chain has lighter blocks, once we've re-orged onto it, we
     # have a greater block height. If the reorg chain has heavier blocks, we
@@ -2639,7 +2639,7 @@ async def test_long_reorg(
     assert peak.header_hash != chain_2_peak
     assert peak.weight > chain_2_weight
 
-    await validate_coin_set(node.full_node._coin_store, blocks)
+    await validate_coin_set(node.full_node.blockchain.consensus_store, blocks)
 
 
 @pytest.mark.anyio
@@ -2665,9 +2665,9 @@ async def test_long_reorg_nodes(
 ) -> None:
     full_node_1, full_node_2, full_node_3 = three_nodes
 
-    assert full_node_1.full_node._coin_store is not None
-    assert full_node_2.full_node._coin_store is not None
-    assert full_node_3.full_node._coin_store is not None
+    assert full_node_1.full_node.blockchain.consensus_store is not None
+    assert full_node_2.full_node.blockchain.consensus_store is not None
+    assert full_node_3.full_node.blockchain.consensus_store is not None
 
     if light_blocks:
         if fork_point == 1500:
@@ -2724,8 +2724,8 @@ def check_nodes_in_sync() -> bool:
     assert p2 is not None
     assert p2.header_hash == reorg_blocks[-1].header_hash
 
-    await validate_coin_set(full_node_1.full_node._coin_store, reorg_blocks)
-    await validate_coin_set(full_node_2.full_node._coin_store, reorg_blocks)
+    await validate_coin_set(full_node_1.full_node.blockchain.consensus_store, reorg_blocks)
+    await validate_coin_set(full_node_2.full_node.blockchain.consensus_store, reorg_blocks)
 
     blocks = default_10000_blocks[:reorg_height]
 
@@ -2765,9 +2765,9 @@ def check_nodes_in_sync2() -> bool:
     print(f"reorg1 timing: {reorg1_timing:0.2f}s")
     print(f"reorg2 timing: {reorg2_timing:0.2f}s")
 
-    await validate_coin_set(full_node_1.full_node._coin_store, blocks)
-    await validate_coin_set(full_node_2.full_node._coin_store, blocks)
-    await validate_coin_set(full_node_3.full_node._coin_store, blocks)
+    await validate_coin_set(full_node_1.full_node.blockchain.consensus_store, blocks)
+    await validate_coin_set(full_node_2.full_node.blockchain.consensus_store, blocks)
+    await validate_coin_set(full_node_3.full_node.blockchain.consensus_store, blocks)
 
 
 @pytest.mark.anyio
@@ -2804,8 +2804,8 @@ def check_nodes_in_sync() -> bool:
         return p1 == p2
 
     await time_out_assert(10, check_nodes_in_sync)
-    await validate_coin_set(full_node_1.full_node.blockchain.coin_store, chain)
-    await validate_coin_set(full_node_2.full_node.blockchain.coin_store, chain)
+    await validate_coin_set(full_node_1.full_node.blockchain.consensus_store, chain)
+    await validate_coin_set(full_node_2.full_node.blockchain.consensus_store, chain)
 
     # we spend a coin in the next block
     spend_bundle = wallet_a.generate_signed_transaction(uint64(1_000), receiver_puzzlehash, all_coins.pop())
@@ -2837,8 +2837,8 @@ def check_nodes_in_sync() -> bool:
     await add_blocks_in_batches(chain_a[-1:], full_node_1.full_node)
 
     await time_out_assert(10, check_nodes_in_sync)
-    await validate_coin_set(full_node_1.full_node.blockchain.coin_store, chain_a)
-    await validate_coin_set(full_node_2.full_node.blockchain.coin_store, chain_a)
+    await validate_coin_set(full_node_1.full_node.blockchain.consensus_store, chain_a)
+    await validate_coin_set(full_node_2.full_node.blockchain.consensus_store, chain_a)
 
     await add_blocks_in_batches(chain_b[-1:], full_node_1.full_node)
 
@@ -2848,8 +2848,8 @@ def check_nodes_in_sync() -> bool:
     assert peak.header_hash == chain_b[-1].header_hash
 
     await time_out_assert(10, check_nodes_in_sync)
-    await validate_coin_set(full_node_1.full_node.blockchain.coin_store, chain_b)
-    await validate_coin_set(full_node_2.full_node.blockchain.coin_store, chain_b)
+    await validate_coin_set(full_node_1.full_node.blockchain.consensus_store, chain_b)
+    await validate_coin_set(full_node_2.full_node.blockchain.consensus_store, chain_b)
 
     # now continue building the chain on top of B
     # since spend_bundle was supposed to have been reorged-out, we should be
@@ -2884,8 +2884,8 @@ def check_nodes_in_sync() -> bool:
 
     await add_blocks_in_batches(chain[-4:], full_node_1.full_node)
     await time_out_assert(10, check_nodes_in_sync)
-    await validate_coin_set(full_node_1.full_node.blockchain.coin_store, chain)
-    await validate_coin_set(full_node_2.full_node.blockchain.coin_store, chain)
+    await validate_coin_set(full_node_1.full_node.blockchain.consensus_store, chain)
+    await validate_coin_set(full_node_2.full_node.blockchain.consensus_store, chain)
 
 
 @pytest.mark.anyio
diff --git a/chia/_tests/core/test_db_conversion.py b/chia/_tests/core/test_db_conversion.py
index d65d9a215e70..bef2ab811e72 100644
--- a/chia/_tests/core/test_db_conversion.py
+++ b/chia/_tests/core/test_db_conversion.py
@@ -15,6 +15,7 @@
 from chia.consensus.multiprocess_validation import PreValidationResult
 from chia.full_node.block_store import BlockStore
 from chia.full_node.coin_store import CoinStore
+from chia.full_node.consensus_store_sqlite3 import ConsensusStoreSQLite3
 from chia.full_node.hint_store import HintStore
 from chia.simulator.block_tools import test_constants
 from chia.util.db_wrapper import DBWrapper2
@@ -62,7 +63,8 @@ async def test_blocks(default_1000_blocks, with_hints: bool):
                     await hint_store1.add_hints([(h[0], h[1])])
 
             height_map = await BlockHeightMap.create(Path("."), db_wrapper1)
-            bc = await Blockchain.create(coin_store1, block_store1, height_map, test_constants, reserved_cores=0)
+            consensus_store = await ConsensusStoreSQLite3.create(block_store1, coin_store1, height_map)
+            bc = await Blockchain.create(consensus_store, test_constants, reserved_cores=0)
             sub_slot_iters = test_constants.SUB_SLOT_ITERS_STARTING
             for block in blocks:
                 if block.height != 0 and len(block.finished_sub_slots) > 0:
diff --git a/chia/_tests/core/test_db_validation.py b/chia/_tests/core/test_db_validation.py
index fef9eb86de10..c26b112f1a61 100644
--- a/chia/_tests/core/test_db_validation.py
+++ b/chia/_tests/core/test_db_validation.py
@@ -19,6 +19,7 @@
 from chia.consensus.multiprocess_validation import PreValidationResult
 from chia.full_node.block_store import BlockStore
 from chia.full_node.coin_store import CoinStore
+from chia.full_node.consensus_store_sqlite3 import ConsensusStoreSQLite3
 from chia.simulator.block_tools import test_constants
 from chia.util.db_wrapper import DBWrapper2
 
@@ -141,8 +142,8 @@ async def make_db(db_file: Path, blocks: list[FullBlock]) -> None:
         block_store = await BlockStore.create(db_wrapper)
         coin_store = await CoinStore.create(db_wrapper)
         height_map = await BlockHeightMap.create(Path("."), db_wrapper)
-
-        bc = await Blockchain.create(coin_store, block_store, height_map, test_constants, reserved_cores=0)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc = await Blockchain.create(consensus_store, test_constants, reserved_cores=0)
         sub_slot_iters = test_constants.SUB_SLOT_ITERS_STARTING
         for block in blocks:
             if block.height != 0 and len(block.finished_sub_slots) > 0:
diff --git a/chia/_tests/util/blockchain.py b/chia/_tests/util/blockchain.py
index 58953062102c..0596e7346904 100644
--- a/chia/_tests/util/blockchain.py
+++ b/chia/_tests/util/blockchain.py
@@ -14,6 +14,7 @@
 from chia.consensus.blockchain import Blockchain
 from chia.full_node.block_store import BlockStore
 from chia.full_node.coin_store import CoinStore
+from chia.full_node.consensus_store_sqlite3 import ConsensusStoreSQLite3
 from chia.simulator.block_tools import BlockTools
 from chia.util.db_wrapper import DBWrapper2, generate_in_memory_db_uri
 from chia.util.default_root import DEFAULT_ROOT_PATH
@@ -25,11 +26,11 @@ async def create_blockchain(
 ) -> AsyncIterator[tuple[Blockchain, DBWrapper2]]:
     db_uri = generate_in_memory_db_uri()
     async with DBWrapper2.managed(database=db_uri, uri=True, reader_count=1, db_version=db_version) as wrapper:
+        block_store = await BlockStore.create(wrapper)
         coin_store = await CoinStore.create(wrapper)
-        store = await BlockStore.create(wrapper)
-        path = Path(".")
-        height_map = await BlockHeightMap.create(path, wrapper)
-        bc1 = await Blockchain.create(coin_store, store, height_map, constants, 3, single_threaded=True, log_coins=True)
+        height_map = await BlockHeightMap.create(Path("."), wrapper, None)
+        consensus_store = await ConsensusStoreSQLite3.create(block_store, coin_store, height_map)
+        bc1 = await Blockchain.create(consensus_store, constants, 3, single_threaded=True, log_coins=True)
         try:
             assert bc1.get_peak() is None
             yield bc1, wrapper
diff --git a/chia/_tests/util/coin_store.py b/chia/_tests/util/coin_store.py
index b5eb266ec825..6a0404923b22 100644
--- a/chia/_tests/util/coin_store.py
+++ b/chia/_tests/util/coin_store.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from chia.consensus.coin_store_protocol import CoinStoreProtocol
+from chia.full_node.coin_store_protocol import CoinStoreProtocol
 from chia.types.coin_record import CoinRecord
 from chia.util.db_wrapper import DBWrapper2
 
diff --git a/chia/consensus/blockchain.py b/chia/consensus/blockchain.py
index 00d9df5d1a4c..731a157e7ff4 100644
--- a/chia/consensus/blockchain.py
+++ b/chia/consensus/blockchain.py
@@ -26,8 +26,7 @@
 
 from chia.consensus.block_body_validation import ForkInfo, validate_block_body
 from chia.consensus.block_header_validation import validate_unfinished_header_block
-from chia.consensus.block_height_map import BlockHeightMap
-from chia.consensus.coin_store_protocol import CoinStoreProtocol
+from chia.consensus.consensus_store_protocol import ConsensusStoreProtocol
 from chia.consensus.cost_calculator import NPCResult
 from chia.consensus.difficulty_adjustment import get_next_sub_slot_iters_and_difficulty
 from chia.consensus.find_fork_point import lookup_fork_chain
@@ -35,7 +34,6 @@
 from chia.consensus.generator_tools import get_block_header
 from chia.consensus.get_block_generator import get_block_generator
 from chia.consensus.multiprocess_validation import PreValidationResult
-from chia.full_node.block_store import BlockStore
 from chia.types.blockchain_format.coin import Coin
 from chia.types.blockchain_format.vdf import VDFInfo
 from chia.types.coin_record import CoinRecord
@@ -100,11 +98,7 @@ class Blockchain:
     __heights_in_cache: dict[uint32, set[bytes32]]
     # maps block height (of the current heaviest chain) to block hash and sub
     # epoch summaries
-    __height_map: BlockHeightMap
-    # Unspent Store
-    coin_store: CoinStoreProtocol
-    # Store
-    block_store: BlockStore
+    consensus_store: ConsensusStoreProtocol
     # Used to verify blocks in parallel
     pool: Executor
     # Set holding seen compact proofs, in order to avoid duplicates.
@@ -121,9 +115,7 @@ class Blockchain:
 
     @staticmethod
     async def create(
-        coin_store: CoinStoreProtocol,
-        block_store: BlockStore,
-        height_map: BlockHeightMap,
+        consensus_store: ConsensusStoreProtocol,
         consensus_constants: ConsensusConstants,
         reserved_cores: int,
         *,
@@ -153,10 +145,9 @@ async def create(
             log.info(f"Started {num_workers} processes for block validation")
 
         self.constants = consensus_constants
-        self.coin_store = coin_store
-        self.block_store = block_store
+        self.consensus_store = consensus_store
         self._shut_down = False
-        await self._load_chain_from_store(height_map)
+        await self._load_chain_from_store()
         self._seen_compact_proofs = set()
         return self
 
@@ -164,14 +155,15 @@ def shut_down(self) -> None:
         self._shut_down = True
         self.pool.shutdown(wait=True)
 
-    async def _load_chain_from_store(self, height_map: BlockHeightMap) -> None:
+    async def _load_chain_from_store(self) -> None:
         """
         Initializes the state of the Blockchain class from the database.
         """
-        self.__height_map = height_map
         self.__block_records = {}
         self.__heights_in_cache = {}
-        block_records, peak = await self.block_store.get_block_records_close_to_peak(self.constants.BLOCKS_CACHE_SIZE)
+        block_records, peak = await self.consensus_store.get_block_records_close_to_peak(
+            self.constants.BLOCKS_CACHE_SIZE
+        )
         for block in block_records.values():
             self.add_block_record(block)
 
@@ -182,8 +174,8 @@ async def _load_chain_from_store(self, height_map: BlockHeightMap) -> None:
 
         assert peak is not None
         self._peak_height = self.block_record(peak).height
-        assert self.__height_map.contains_height(self._peak_height)
-        assert not self.__height_map.contains_height(uint32(self._peak_height + 1))
+        assert self.consensus_store.contains_height(self._peak_height)
+        assert not self.consensus_store.contains_height(uint32(self._peak_height + 1))
 
     def get_peak(self) -> Optional[BlockRecord]:
         """
@@ -219,12 +211,12 @@ async def get_full_peak(self) -> Optional[FullBlock]:
         """ Return list of FullBlocks that are peaks"""
         peak_hash: Optional[bytes32] = self.height_to_hash(self._peak_height)
         assert peak_hash is not None  # Since we must have the peak block
-        block = await self.block_store.get_full_block(peak_hash)
+        block = await self.consensus_store.get_full_block(peak_hash)
         assert block is not None
         return block
 
     async def get_full_block(self, header_hash: bytes32) -> Optional[FullBlock]:
-        return await self.block_store.get_full_block(header_hash)
+        return await self.consensus_store.get_full_block(header_hash)
 
     async def advance_fork_info(self, block: FullBlock, fork_info: ForkInfo) -> None:
         """
@@ -258,7 +250,7 @@ async def advance_fork_info(self, block: FullBlock, fork_info: ForkInfo) -> None
         assert len(chain) == block.height - fork_info.peak_height - 1
 
         for height in range(fork_info.peak_height + 1, block.height):
-            fork_block: Optional[FullBlock] = await self.block_store.get_full_block(chain[uint32(height)])
+            fork_block: Optional[FullBlock] = await self.consensus_store.get_full_block(chain[uint32(height)])
             assert fork_block is not None
             await self.run_single_block(fork_block, fork_info)
 
@@ -384,7 +376,7 @@ async def add_block(
         error_code = await validate_block_body(
             self.constants,
             self,
-            self.coin_store.get_coin_records,
+            self.consensus_store.get_coin_records,
             block,
             block.height,
             pre_validation_result.conds,
@@ -421,9 +413,9 @@ async def add_block(
 
         try:
             # Always add the block to the database
-            async with self.block_store.transaction():
+            async with self.consensus_store as writer:
                 # Perform the DB operations to update the state, and rollback if something goes wrong
-                await self.block_store.add_full_block(header_hash, block, block_record)
+                await writer.add_full_block(header_hash, block, block_record)
                 records, state_change_summary = await self._reconsider_peak(block_record, genesis, fork_info)
 
                 # Then update the memory cache. It is important that this is not cancelled and does not throw
@@ -436,9 +428,9 @@ async def add_block(
             # make sure to update _peak_height after the transaction is committed,
             # otherwise other tasks may go look for this block before it's available
             if state_change_summary is not None:
-                self.__height_map.rollback(state_change_summary.fork_height)
+                self.consensus_store.rollback_height_map(state_change_summary.fork_height)
             for fetched_block_record in records:
-                self.__height_map.update_height(
+                self.consensus_store.update_height_map(
                     fetched_block_record.height,
                     fetched_block_record.header_hash,
                     fetched_block_record.sub_epoch_summary_included,
@@ -456,7 +448,7 @@ async def add_block(
                 pass
             # restore fork_info to the state before adding the block
             fork_info.rollback(prev_fork_peak[1], prev_fork_peak[0])
-            self.block_store.rollback_cache_block(header_hash)
+            self.consensus_store.rollback_cache_block(header_hash)
             self._peak_height = previous_peak_height
             log.error(
                 f"Error while adding block {header_hash} height {block.height},"
@@ -465,7 +457,7 @@ async def add_block(
             raise
 
         # This is done outside the try-except in case it fails, since we do not want to revert anything if it does
-        await self.__height_map.maybe_flush()
+        await self.consensus_store.maybe_flush_height_map()
 
         if state_change_summary is not None:
             # new coin records added
@@ -494,108 +486,109 @@ async def _reconsider_peak(
         if genesis and peak is not None:
             return [], None
 
-        if peak is not None:
-            if block_record.weight < peak.weight:
-                # This is not a heavier block than the heaviest we have seen, so we don't change the coin set
-                return [], None
-            if block_record.weight == peak.weight and peak.total_iters <= block_record.total_iters:
-                # this is an equal weight block but our peak has lower iterations, so we dont change the coin set
-                return [], None
-            if block_record.weight == peak.weight:
-                log.info(
-                    f"block has equal weight as our peak ({peak.weight}), but fewer "
-                    f"total iterations {block_record.total_iters} "
-                    f"peak: {peak.total_iters} "
-                    f"peak-hash: {peak.header_hash}"
-                )
-
-            if block_record.prev_hash != peak.header_hash:
-                rolled_back_state = await self.coin_store.rollback_to_block(fork_info.fork_height)
-                if self._log_coins and len(rolled_back_state) > 0:
-                    log.info(f"rolled back {len(rolled_back_state)} coins, to fork height {fork_info.fork_height}")
+        async with self.consensus_store as writer:
+            if peak is not None:
+                if block_record.weight < peak.weight:
+                    # This is not a heavier block than the heaviest we have seen, so we don't change the coin set
+                    return [], None
+                if block_record.weight == peak.weight and peak.total_iters <= block_record.total_iters:
+                    # this is an equal weight block but our peak has lower iterations, so we dont change the coin set
+                    return [], None
+                if block_record.weight == peak.weight:
                     log.info(
-                        "removed: %s",
-                        ",".join(
-                            [
-                                name.hex()[0:6]
-                                for name, state in rolled_back_state.items()
-                                if state.confirmed_block_index == 0
-                            ]
-                        ),
+                        f"block has equal weight as our peak ({peak.weight}), but fewer "
+                        f"total iterations {block_record.total_iters} "
+                        f"peak: {peak.total_iters} "
+                        f"peak-hash: {peak.header_hash}"
                     )
-                    log.info(
-                        "unspent: %s",
-                        ",".join(
-                            [
-                                name.hex()[0:6]
-                                for name, state in rolled_back_state.items()
-                                if state.confirmed_block_index != 0
-                            ]
-                        ),
-                    )
-
-        # Collects all blocks from fork point to new peak
-        records_to_add: list[BlockRecord] = []
 
-        if genesis:
-            records_to_add = [block_record]
-        elif fork_info.block_hashes == [block_record.header_hash]:
-            # in the common case, we just add a block on top of the chain. Check
-            # for that here to avoid an unnecessary database lookup.
-            records_to_add = [block_record]
-        else:
-            records_to_add = await self.block_store.get_block_records_by_hash(fork_info.block_hashes)
-
-        for fetched_block_record in records_to_add:
-            if not fetched_block_record.is_transaction_block:
-                # Coins are only created in TX blocks so there are no state updates for this block
-                continue
-
-            height = fetched_block_record.height
-            # We need to recompute the additions and removals, since they are
-            # not stored on DB. We have all the additions and removals in the
-            # fork_info object, we just need to pick the ones belonging to each
-            # individual block height
-
-            # Apply the coin store changes for each block that is now in the blockchain
-            included_reward_coins = [
-                fork_add.coin
-                for fork_add in fork_info.additions_since_fork.values()
-                if fork_add.confirmed_height == height and fork_add.is_coinbase
-            ]
-            tx_additions = [
-                (coin_id, fork_add.coin, fork_add.same_as_parent)
-                for coin_id, fork_add in fork_info.additions_since_fork.items()
-                if fork_add.confirmed_height == height and not fork_add.is_coinbase
-            ]
-            tx_removals = [
-                coin_id for coin_id, fork_rem in fork_info.removals_since_fork.items() if fork_rem.height == height
-            ]
-            assert fetched_block_record.timestamp is not None
-            await self.coin_store.new_block(
-                height,
-                fetched_block_record.timestamp,
-                included_reward_coins,
-                tx_additions,
-                tx_removals,
-            )
-            if self._log_coins and (len(tx_removals) > 0 or len(tx_additions) > 0):
-                log.info(
-                    f"adding new block to coin_store "
-                    f"(hh: {fetched_block_record.header_hash} "
-                    f"height: {fetched_block_record.height}), {len(tx_removals)} spends"
+                if block_record.prev_hash != peak.header_hash:
+                    rolled_back_state = await writer.rollback_to_block(fork_info.fork_height)
+                    if self._log_coins and len(rolled_back_state) > 0:
+                        log.info(f"rolled back {len(rolled_back_state)} coins, to fork height {fork_info.fork_height}")
+                        log.info(
+                            "removed: %s",
+                            ",".join(
+                                [
+                                    name.hex()[0:6]
+                                    for name, state in rolled_back_state.items()
+                                    if state.confirmed_block_index == 0
+                                ]
+                            ),
+                        )
+                        log.info(
+                            "unspent: %s",
+                            ",".join(
+                                [
+                                    name.hex()[0:6]
+                                    for name, state in rolled_back_state.items()
+                                    if state.confirmed_block_index != 0
+                                ]
+                            ),
+                        )
+
+            # Collects all blocks from fork point to new peak
+            records_to_add: list[BlockRecord] = []
+
+            if genesis:
+                records_to_add = [block_record]
+            elif fork_info.block_hashes == [block_record.header_hash]:
+                # in the common case, we just add a block on top of the chain. Check
+                # for that here to avoid an unnecessary database lookup.
+                records_to_add = [block_record]
+            else:
+                records_to_add = await self.consensus_store.get_block_records_by_hash(fork_info.block_hashes)
+
+            for fetched_block_record in records_to_add:
+                if not fetched_block_record.is_transaction_block:
+                    # Coins are only created in TX blocks so there are no state updates for this block
+                    continue
+
+                height = fetched_block_record.height
+                # We need to recompute the additions and removals, since they are
+                # not stored on DB. We have all the additions and removals in the
+                # fork_info object, we just need to pick the ones belonging to each
+                # individual block height
+
+                # Apply the coin store changes for each block that is now in the blockchain
+                included_reward_coins = [
+                    fork_add.coin
+                    for fork_add in fork_info.additions_since_fork.values()
+                    if fork_add.confirmed_height == height and fork_add.is_coinbase
+                ]
+                tx_additions = [
+                    (coin_id, fork_add.coin, fork_add.same_as_parent)
+                    for coin_id, fork_add in fork_info.additions_since_fork.items()
+                    if fork_add.confirmed_height == height and not fork_add.is_coinbase
+                ]
+                tx_removals = [
+                    coin_id for coin_id, fork_rem in fork_info.removals_since_fork.items() if fork_rem.height == height
+                ]
+                assert fetched_block_record.timestamp is not None
+                await writer.new_block(
+                    height,
+                    fetched_block_record.timestamp,
+                    included_reward_coins,
+                    tx_additions,
+                    tx_removals,
                 )
-                log.info("rewards: %s", ",".join([add.name().hex()[0:6] for add in included_reward_coins]))
-                log.info("additions: %s", ",".join([add[0].hex()[0:6] for add in tx_additions]))
-                log.info("removals: %s", ",".join([f"{rem}"[0:6] for rem in tx_removals]))
+                if self._log_coins and (len(tx_removals) > 0 or len(tx_additions) > 0):
+                    log.info(
+                        f"adding new block to coin_store "
+                        f"(hh: {fetched_block_record.header_hash} "
+                        f"height: {fetched_block_record.height}), {len(tx_removals)} spends"
+                    )
+                    log.info("rewards: %s", ",".join([add.name().hex()[0:6] for add in included_reward_coins]))
+                    log.info("additions: %s", ",".join([add[0].hex()[0:6] for add in tx_additions]))
+                    log.info("removals: %s", ",".join([f"{rem}"[0:6] for rem in tx_removals]))
 
-        # we made it to the end successfully
-        # Rollback sub_epoch_summaries
-        await self.block_store.rollback(fork_info.fork_height)
-        await self.block_store.set_in_chain([(br.header_hash,) for br in records_to_add])
+            # we made it to the end successfully
+            # Rollback sub_epoch_summaries
+            await writer.rollback(fork_info.fork_height)
+            await writer.set_in_chain([(br.header_hash,) for br in records_to_add])
 
-        # Changes the peak to be the new peak
-        await self.block_store.set_peak(block_record.header_hash)
+            # Changes the peak to be the new peak
+            await writer.set_peak(block_record.header_hash)
 
         return records_to_add, StateChangeSummary(
             block_record,
@@ -621,7 +614,7 @@ def get_next_sub_slot_iters_and_difficulty(self, header_hash: bytes32, new_slot:
     async def get_sp_and_ip_sub_slots(
         self, header_hash: bytes32
     ) -> Optional[tuple[Optional[EndOfSubSlotBundle], Optional[EndOfSubSlotBundle]]]:
-        block: Optional[FullBlock] = await self.block_store.get_full_block(header_hash)
+        block: Optional[FullBlock] = await self.consensus_store.get_full_block(header_hash)
         if block is None:
             return None
         curr_br: BlockRecord = self.block_record(block.header_hash)
@@ -631,7 +624,7 @@ async def get_sp_and_ip_sub_slots(
         assert curr is not None
         while True:
             if curr_br.first_in_sub_slot:
-                curr = await self.block_store.get_full_block(curr_br.header_hash)
+                curr = await self.consensus_store.get_full_block(curr_br.header_hash)
                 assert curr is not None
                 break
             if curr_br.height == 0:
@@ -652,7 +645,7 @@ async def get_sp_and_ip_sub_slots(
             # Have both sub-slots
             return curr.finished_sub_slots[-2], ip_sub_slot
 
-        prev_curr: Optional[FullBlock] = await self.block_store.get_full_block(curr.prev_header_hash)
+        prev_curr: Optional[FullBlock] = await self.consensus_store.get_full_block(curr.prev_header_hash)
         if prev_curr is None:
             assert curr.height == 0
             prev_curr = curr
@@ -662,7 +655,7 @@ async def get_sp_and_ip_sub_slots(
         assert prev_curr_br is not None
         while prev_curr_br.height > 0:
             if prev_curr_br.first_in_sub_slot:
-                prev_curr = await self.block_store.get_full_block(prev_curr_br.header_hash)
+                prev_curr = await self.consensus_store.get_full_block(prev_curr_br.header_hash)
                 assert prev_curr is not None
                 break
             prev_curr_br = self.block_record(prev_curr_br.prev_hash)
@@ -769,7 +762,7 @@ async def validate_unfinished_block(
         error_code = await validate_block_body(
             self.constants,
             self,
-            self.coin_store.get_coin_records,
+            self.consensus_store.get_coin_records,
             block,
             uint32(prev_height + 1),
             conds,
@@ -799,18 +792,18 @@ def height_to_block_record(self, height: uint32) -> BlockRecord:
         return self.block_record(header_hash)
 
     def get_ses_heights(self) -> list[uint32]:
-        return self.__height_map.get_ses_heights()
+        return self.consensus_store.get_ses_heights()
 
     def get_ses(self, height: uint32) -> SubEpochSummary:
-        return self.__height_map.get_ses(height)
+        return self.consensus_store.get_ses(height)
 
     def height_to_hash(self, height: uint32) -> Optional[bytes32]:
-        if not self.__height_map.contains_height(height):
+        if not self.consensus_store.contains_height(height):
             return None
-        return self.__height_map.get_hash(height)
+        return self.consensus_store.get_hash(height)
 
     def contains_height(self, height: uint32) -> bool:
-        return self.__height_map.contains_height(height)
+        return self.consensus_store.contains_height(height)
 
     def get_peak_height(self) -> Optional[uint32]:
         return self._peak_height
@@ -826,7 +819,7 @@ async def warmup(self, fork_point: uint32) -> None:
         """
         if self._peak_height is None:
             return None
-        block_records = await self.block_store.get_block_records_in_range(
+        block_records = await self.consensus_store.get_block_records_in_range(
             max(fork_point - self.constants.BLOCKS_CACHE_SIZE, uint32(0)), fork_point
         )
         for block_record in block_records.values():
@@ -869,7 +862,7 @@ def clean_block_records(self) -> None:
         self.clean_block_record(self._peak_height - self.constants.BLOCKS_CACHE_SIZE)
 
     async def get_block_records_in_range(self, start: int, stop: int) -> dict[bytes32, BlockRecord]:
-        return await self.block_store.get_block_records_in_range(start, stop)
+        return await self.consensus_store.get_block_records_in_range(start, stop)
 
     async def get_header_blocks_in_range(
         self, start: int, stop: int, tx_filter: bool = True
@@ -882,11 +875,11 @@ async def get_header_blocks_in_range(
 
         blocks: list[FullBlock] = []
         for hash in hashes.copy():
-            block = self.block_store.get_block_from_cache(hash)
+            block = self.consensus_store.get_block_from_cache(hash)
             if block is not None:
                 blocks.append(block)
                 hashes.remove(hash)
-        blocks_on_disk: list[FullBlock] = await self.block_store.get_blocks_by_hash(hashes)
+        blocks_on_disk: list[FullBlock] = await self.consensus_store.get_blocks_by_hash(hashes)
         blocks.extend(blocks_on_disk)
         header_blocks: dict[bytes32, HeaderBlock] = {}
 
@@ -897,8 +890,8 @@ async def get_header_blocks_in_range(
                 header = get_block_header(block)
             elif block.transactions_generator is not None:
                 added_coins_records, removed_coins_records = await asyncio.gather(
-                    self.coin_store.get_coins_added_at_height(block.height),
-                    self.coin_store.get_coins_removed_at_height(block.height),
+                    self.consensus_store.get_coins_added_at_height(block.height),
+                    self.consensus_store.get_coins_removed_at_height(block.height),
                 )
                 tx_additions = [cr.coin for cr in added_coins_records if not cr.coinbase]
                 removed = [cr.coin.name() for cr in removed_coins_records]
@@ -936,7 +929,7 @@ async def get_block_records_at(self, heights: list[uint32]) -> list[BlockRecord]
                 raise ValueError(f"Do not have block at height {height}")
             hashes.append(header_hash)
 
-        return await self.block_store.get_block_records_by_hash(hashes)
+        return await self.consensus_store.get_block_records_by_hash(hashes)
 
     def try_block_record(self, header_hash: bytes32) -> Optional[BlockRecord]:
         if header_hash in self.__block_records:
@@ -947,7 +940,7 @@ async def get_block_record_from_db(self, header_hash: bytes32) -> Optional[Block
         ret = self.__block_records.get(header_hash)
         if ret is not None:
             return ret
-        return await self.block_store.get_block_record(header_hash)
+        return await self.consensus_store.get_block_record(header_hash)
 
     async def prev_block_hash(self, header_hashes: list[bytes32]) -> list[bytes32]:
         """
@@ -960,7 +953,7 @@ async def prev_block_hash(self, header_hashes: list[bytes32]) -> list[bytes32]:
             if b is not None:
                 ret.append(b.prev_hash)
             else:
-                ret.append(await self.block_store.get_prev_hash(h))
+                ret.append(await self.consensus_store.get_prev_hash(h))
         return ret
 
     async def contains_block_from_db(self, header_hash: bytes32) -> bool:
@@ -968,7 +961,7 @@ async def contains_block_from_db(self, header_hash: bytes32) -> bool:
         if ret:
             return True
 
-        return (await self.block_store.get_block_record(header_hash)) is not None
+        return (await self.consensus_store.get_block_record(header_hash)) is not None
 
     def remove_block_record(self, header_hash: bytes32) -> None:
         sbr = self.block_record(header_hash)
@@ -988,15 +981,16 @@ def add_block_record(self, block_record: BlockRecord) -> None:
     async def persist_sub_epoch_challenge_segments(
         self, ses_block_hash: bytes32, segments: list[SubEpochChallengeSegment]
     ) -> None:
-        await self.block_store.persist_sub_epoch_challenge_segments(ses_block_hash, segments)
+        async with self.consensus_store as writer:
+            await writer.persist_sub_epoch_challenge_segments(ses_block_hash, segments)
 
     async def get_sub_epoch_challenge_segments(
         self,
         ses_block_hash: bytes32,
     ) -> Optional[list[SubEpochChallengeSegment]]:
-        segments: Optional[list[SubEpochChallengeSegment]] = await self.block_store.get_sub_epoch_challenge_segments(
-            ses_block_hash
-        )
+        segments: Optional[
+            list[SubEpochChallengeSegment]
+        ] = await self.consensus_store.get_sub_epoch_challenge_segments(ses_block_hash)
         if segments is None:
             return None
         return segments
@@ -1054,7 +1048,7 @@ async def lookup_block_generators(self, header_hash: bytes32, generator_refs: se
             remaining_refs = set()
             for ref_height in generator_refs:
                 if ref_height in reorg_chain:
-                    gen = await self.block_store.get_generator(reorg_chain[ref_height])
+                    gen = await self.consensus_store.get_generator(reorg_chain[ref_height])
                     if gen is None:
                         raise ValueError(Err.GENERATOR_REF_HAS_NO_GENERATOR)
                     generators[ref_height] = gen
@@ -1066,6 +1060,6 @@ async def lookup_block_generators(self, header_hash: bytes32, generator_refs: se
         if len(remaining_refs) > 0:
             # any remaining references fall in the main chain, and can be looked up
             # in a single query
-            generators.update(await self.block_store.get_generators_at(remaining_refs))
+            generators.update(await self.consensus_store.get_generators_at(remaining_refs))
 
         return generators
diff --git a/chia/consensus/consensus_store_protocol.py b/chia/consensus/consensus_store_protocol.py
new file mode 100644
index 000000000000..459df7f9565c
--- /dev/null
+++ b/chia/consensus/consensus_store_protocol.py
@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Collection
+from types import TracebackType
+from typing import Optional, Protocol
+
+from chia_rs import BlockRecord, FullBlock, SubEpochChallengeSegment, SubEpochSummary
+from chia_rs.sized_bytes import bytes32
+from chia_rs.sized_ints import uint32, uint64
+
+from chia.types.blockchain_format.coin import Coin
+from chia.types.coin_record import CoinRecord
+
+
+class ConsensusStoreWriteProtocol(Protocol):
+    """
+    Protocol for performing mutating operations on the consensus store.
+
+    Instances implementing this protocol should be acquired via the async
+    context manager on ConsensusStoreProtocol to ensure atomic write
+    operations (e.g., wrapping all writes in a single DB transaction).
+    """
+
+    # Block store writes
+    async def add_full_block(self, header_hash: bytes32, block: FullBlock, block_record: BlockRecord) -> None: ...
+    async def rollback(self, height: int) -> None: ...
+    async def set_in_chain(self, header_hashes: list[tuple[bytes32]]) -> None: ...
+    async def set_peak(self, header_hash: bytes32) -> None: ...
+    async def persist_sub_epoch_challenge_segments(
+        self, ses_block_hash: bytes32, segments: list[SubEpochChallengeSegment]
+    ) -> None: ...
+
+    # Coin store writes
+    async def rollback_to_block(self, block_index: int) -> dict[bytes32, CoinRecord]: ...
+    async def new_block(
+        self,
+        height: uint32,
+        timestamp: uint64,
+        included_reward_coins: Collection[Coin],
+        tx_additions: Collection[tuple[bytes32, Coin, bool]],
+        tx_removals: list[bytes32],
+    ) -> None: ...
+
+
+class ConsensusStoreProtocol(Protocol):
+    """
+    Read-only protocol for the consensus store.
+
+    This protocol also acts as an async context manager. Entering the context
+    yields a ConsensusStoreWriteProtocol instance, which must be used for
+    performing write (mutating) operations. This ensures atomic writes and
+    makes it harder to accidentally perform writes outside a transaction.
+
+    Example usage:
+        async with store as writer:
+            await writer.add_full_block(...)
+            await writer.set_peak(...)
+
+        # Outside the context, only read methods are available
+        br = await store.get_block_record(header_hash)
+    """
+
+    # Async context manager methods
+    async def __aenter__(self) -> ConsensusStoreWriteProtocol: ...
+    async def __aexit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc: Optional[BaseException],
+        tb: Optional[TracebackType],
+    ) -> Optional[bool]: ...
+
+    # Block store reads
+    async def get_block_records_close_to_peak(
+        self, blocks_n: int
+    ) -> tuple[dict[bytes32, BlockRecord], Optional[bytes32]]: ...
+    async def get_full_block(self, header_hash: bytes32) -> Optional[FullBlock]: ...
+    async def get_block_records_by_hash(self, header_hashes: list[bytes32]) -> list[BlockRecord]: ...
+    async def get_block_records_in_range(self, start: int, stop: int) -> dict[bytes32, BlockRecord]: ...
+    def get_block_from_cache(self, header_hash: bytes32) -> Optional[FullBlock]: ...
+    async def get_blocks_by_hash(self, header_hashes: list[bytes32]) -> list[FullBlock]: ...
+    async def get_block_record(self, header_hash: bytes32) -> Optional[BlockRecord]: ...
+    async def get_prev_hash(self, header_hash: bytes32) -> bytes32: ...
+    async def get_sub_epoch_challenge_segments(
+        self, ses_block_hash: bytes32
+    ) -> Optional[list[SubEpochChallengeSegment]]: ...
+    async def get_generator(self, header_hash: bytes32) -> Optional[bytes]: ...
+    async def get_generators_at(self, heights: set[uint32]) -> dict[uint32, bytes]: ...
+    def get_block_heights_in_main_chain(self) -> AsyncIterator[int]: ...
+
+    # Coin store reads
+    async def get_coin_records(self, names: Collection[bytes32]) -> list[CoinRecord]: ...
+    async def get_coins_added_at_height(self, height: uint32) -> list[CoinRecord]: ...
+    async def get_coins_removed_at_height(self, height: uint32) -> list[CoinRecord]: ...
+
+    # Height map methods (kept here for now; non-async and maybe_flush remain on read protocol)
+    def get_ses_heights(self) -> list[uint32]: ...
+    def get_ses(self, height: uint32) -> SubEpochSummary: ...
+    def contains_height(self, height: uint32) -> bool: ...
+    def get_hash(self, height: uint32) -> bytes32: ...
+    def rollback_height_map(self, height: uint32) -> None: ...
+    def update_height_map(self, height: uint32, block_hash: bytes32, ses: Optional[SubEpochSummary]) -> None: ...
+    async def maybe_flush_height_map(self) -> None: ...
+    def rollback_cache_block(self, header_hash: bytes32) -> None: ...
diff --git a/chia/consensus/coin_store_protocol.py b/chia/full_node/coin_store_protocol.py
similarity index 100%
rename from chia/consensus/coin_store_protocol.py
rename to chia/full_node/coin_store_protocol.py
diff --git a/chia/full_node/consensus_store_sqlite3.py b/chia/full_node/consensus_store_sqlite3.py
new file mode 100644
index 000000000000..7cf8e44960a1
--- /dev/null
+++ b/chia/full_node/consensus_store_sqlite3.py
@@ -0,0 +1,220 @@
+from __future__ import annotations
+
+import dataclasses
+from collections.abc import AsyncIterator, Collection
+from contextlib import AbstractAsyncContextManager
+from types import TracebackType
+from typing import TYPE_CHECKING, Any, Optional
+
+from chia_rs import BlockRecord, FullBlock, SubEpochChallengeSegment, SubEpochSummary
+from chia_rs.sized_bytes import bytes32
+from chia_rs.sized_ints import uint32, uint64
+
+from chia.consensus.block_height_map import BlockHeightMap
+from chia.full_node.block_store import BlockStore
+from chia.full_node.coin_store import CoinStore
+from chia.types.blockchain_format.coin import Coin
+from chia.types.coin_record import CoinRecord
+
+
+class ConsensusStoreSQLite3Writer:
+    def __init__(self, block_store: BlockStore, coin_store: CoinStore):
+        self._block_store = block_store
+        self._coin_store = coin_store
+
+    async def add_full_block(self, header_hash: bytes32, block: FullBlock, block_record: BlockRecord) -> None:
+        await self._block_store.add_full_block(header_hash, block, block_record)
+
+    async def rollback(self, height: int) -> None:
+        await self._block_store.rollback(height)
+
+    async def set_in_chain(self, header_hashes: list[tuple[bytes32]]) -> None:
+        await self._block_store.set_in_chain(header_hashes)
+
+    async def set_peak(self, header_hash: bytes32) -> None:
+        await self._block_store.set_peak(header_hash)
+
+    async def persist_sub_epoch_challenge_segments(
+        self, ses_block_hash: bytes32, segments: list[SubEpochChallengeSegment]
+    ) -> None:
+        await self._block_store.persist_sub_epoch_challenge_segments(ses_block_hash, segments)
+
+    async def rollback_to_block(self, block_index: int) -> dict[bytes32, CoinRecord]:
+        return await self._coin_store.rollback_to_block(block_index)
+
+    async def new_block(
+        self,
+        height: uint32,
+        timestamp: uint64,
+        included_reward_coins: Collection[Coin],
+        tx_additions: Collection[tuple[bytes32, Coin, bool]],
+        tx_removals: list[bytes32],
+    ) -> None:
+        await self._coin_store.new_block(height, timestamp, included_reward_coins, tx_additions, tx_removals)
+
+
+@dataclasses.dataclass
+class ConsensusStoreSQLite3:
+    """
+    Consensus store that combines block_store, coin_store, and height_map functionality.
+    """
+
+    block_store: BlockStore
+    coin_store: CoinStore
+    height_map: BlockHeightMap
+
+    # Writer context and writer facade for transactional writes (re-entrant via depth counter)
+    _writer_ctx: Optional[AbstractAsyncContextManager[Any]] = None
+    _writer: Optional[ConsensusStoreSQLite3Writer] = None
+    _txn_depth: int = 0
+
+    @classmethod
+    async def create(
+        cls,
+        block_store: BlockStore,
+        coin_store: CoinStore,
+        height_map: BlockHeightMap,
+    ) -> ConsensusStoreSQLite3:
+        """Create a new ConsensusStore instance from existing sub-stores.
+
+        This factory does not create sub-stores. Construct BlockStore, CoinStore,
+        and BlockHeightMap separately and pass them in here.
+        """
+        return cls(
+            block_store=block_store,
+            coin_store=coin_store,
+            height_map=height_map,
+        )
+
+    # Async context manager yielding a writer for atomic writes
+    async def __aenter__(self) -> ConsensusStoreSQLite3Writer:
+        # Re-entrant async context manager:
+        # Begin a transaction only at the outermost level. CoinStore shares the same DB.
+        if self._writer is None:
+            self._writer_ctx = self.block_store.transaction()
+            await self._writer_ctx.__aenter__()
+            # Create writer facade bound to this transaction
+            self._writer = ConsensusStoreSQLite3Writer(self.block_store, self.coin_store)
+        self._txn_depth += 1
+        return self._writer  # Return the same writer for nested contexts
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc: Optional[BaseException],
+        tb: Optional[TracebackType],
+    ) -> Optional[bool]:
+        try:
+            # Check if we're at the outermost level before decrementing
+            if self._txn_depth == 1:
+                # This is the outermost context, handle transaction exit
+                if self._writer_ctx is not None:
+                    return await self._writer_ctx.__aexit__(exc_type, exc, tb)
+                return None
+            else:
+                # This is a nested context, just return None (don't suppress exceptions)
+                return None
+        finally:
+            # Always decrement depth and clean up if we're at the outermost level
+            if self._txn_depth > 0:
+                self._txn_depth -= 1
+            if self._txn_depth == 0:
+                self._writer_ctx = None
+                self._writer = None
+
+    # Block store methods
+
+    async def get_block_records_close_to_peak(
+        self, blocks_n: int
+    ) -> tuple[dict[bytes32, BlockRecord], Optional[bytes32]]:
+        return await self.block_store.get_block_records_close_to_peak(blocks_n)
+
+    async def get_full_block(self, header_hash: bytes32) -> Optional[FullBlock]:
+        return await self.block_store.get_full_block(header_hash)
+
+    async def get_block_records_by_hash(self, header_hashes: list[bytes32]) -> list[BlockRecord]:
+        return await self.block_store.get_block_records_by_hash(header_hashes)
+
+    async def get_block_records_in_range(self, start: int, stop: int) -> dict[bytes32, BlockRecord]:
+        return await self.block_store.get_block_records_in_range(start, stop)
+
+    def get_block_from_cache(self, header_hash: bytes32) -> Optional[FullBlock]:
+        return self.block_store.get_block_from_cache(header_hash)
+
+    async def get_blocks_by_hash(self, header_hashes: list[bytes32]) -> list[FullBlock]:
+        return await self.block_store.get_blocks_by_hash(header_hashes)
+
+    async def get_block_record(self, header_hash: bytes32) -> Optional[BlockRecord]:
+        return await self.block_store.get_block_record(header_hash)
+
+    async def get_prev_hash(self, header_hash: bytes32) -> bytes32:
+        return await self.block_store.get_prev_hash(header_hash)
+
+    async def get_sub_epoch_challenge_segments(
+        self, ses_block_hash: bytes32
+    ) -> Optional[list[SubEpochChallengeSegment]]:
+        return await self.block_store.get_sub_epoch_challenge_segments(ses_block_hash)
+
+    async def get_generator(self, header_hash: bytes32) -> Optional[bytes]:
+        return await self.block_store.get_generator(header_hash)
+
+    async def get_generators_at(self, heights: set[uint32]) -> dict[uint32, bytes]:
+        return await self.block_store.get_generators_at(heights)
+
+    # Coin store methods
+    async def get_coin_records(self, names: Collection[bytes32]) -> list[CoinRecord]:
+        return await self.coin_store.get_coin_records(names)
+
+    async def get_coins_added_at_height(self, height: uint32) -> list[CoinRecord]:
+        return await self.coin_store.get_coins_added_at_height(height)
+
+    async def get_coins_removed_at_height(self, height: uint32) -> list[CoinRecord]:
+        return await self.coin_store.get_coins_removed_at_height(height)
+
+    def get_block_heights_in_main_chain(self) -> AsyncIterator[int]:
+        async def gen() -> AsyncIterator[int]:
+            async with self.block_store.transaction() as conn:
+                async with conn.execute("SELECT height, in_main_chain FROM full_blocks") as cursor:
+                    async for row in cursor:
+                        if row[1]:
+                            yield row[0]
+
+        return gen()
+
+    # Height map methods
+    def get_ses_heights(self) -> list[uint32]:
+        return self.height_map.get_ses_heights()
+
+    def get_ses(self, height: uint32) -> SubEpochSummary:
+        return self.height_map.get_ses(height)
+
+    def contains_height(self, height: uint32) -> bool:
+        return self.height_map.contains_height(height)
+
+    def get_hash(self, height: uint32) -> bytes32:
+        return self.height_map.get_hash(height)
+
+    def rollback_height_map(self, height: uint32) -> None:
+        # BlockHeightMap.rollback is synchronous
+        self.height_map.rollback(height)
+
+    def update_height_map(self, height: uint32, block_hash: bytes32, ses: Optional[SubEpochSummary]) -> None:
+        # BlockHeightMap exposes update_height(height, header_hash, ses)
+        self.height_map.update_height(height, block_hash, ses)
+
+    async def maybe_flush_height_map(self) -> None:
+        # BlockHeightMap.maybe_flush is asynchronous
+        await self.height_map.maybe_flush()
+
+    def rollback_cache_block(self, header_hash: bytes32) -> None:
+        self.block_store.rollback_cache_block(header_hash)
+
+
+if TYPE_CHECKING:
+    from typing import cast
+
+    from chia.consensus.consensus_store_protocol import ConsensusStoreProtocol
+
+    def _protocol_check(o: ConsensusStoreProtocol) -> None: ...
+
+    _protocol_check(cast(ConsensusStoreSQLite3, None))
diff --git a/chia/full_node/full_node.py b/chia/full_node/full_node.py
index 28a8c9c397d2..2b09ec388a9f 100644
--- a/chia/full_node/full_node.py
+++ b/chia/full_node/full_node.py
@@ -42,7 +42,6 @@
 from chia.consensus.block_height_map import BlockHeightMap
 from chia.consensus.blockchain import AddBlockResult, Blockchain, BlockchainMutexPriority, StateChangeSummary
 from chia.consensus.blockchain_interface import BlockchainInterface
-from chia.consensus.coin_store_protocol import CoinStoreProtocol
 from chia.consensus.condition_tools import pkm_pairs
 from chia.consensus.cost_calculator import NPCResult
 from chia.consensus.difficulty_adjustment import get_next_sub_slot_iters_and_difficulty
@@ -53,6 +52,7 @@
 from chia.full_node.block_store import BlockStore
 from chia.full_node.check_fork_next_block import check_fork_next_block
 from chia.full_node.coin_store import CoinStore
+from chia.full_node.consensus_store_sqlite3 import ConsensusStoreSQLite3
 from chia.full_node.full_node_api import FullNodeAPI
 from chia.full_node.full_node_store import FullNodeStore, FullNodeStorePeakResult, UnfinishedBlockEntry
 from chia.full_node.hint_management import get_hints_and_subscription_coin_ids
@@ -157,7 +157,7 @@ class FullNode:
     _db_wrapper: Optional[DBWrapper2] = None
     _hint_store: Optional[HintStore] = None
     _block_store: Optional[BlockStore] = None
-    _coin_store: Optional[CoinStoreProtocol] = None
+    _coin_store: Optional[CoinStore] = None
     _mempool_manager: Optional[MempoolManager] = None
     _init_weight_proof: Optional[asyncio.Task[None]] = None
     _blockchain: Optional[Blockchain] = None
@@ -266,12 +266,11 @@ async def manage(self) -> AsyncIterator[None]:
             multiprocessing_start_method = process_config_start_method(config=self.config, log=self.log)
             self.multiprocessing_context = multiprocessing.get_context(method=multiprocessing_start_method)
             selected_network = self.config.get("selected_network")
-            height_map = await BlockHeightMap.create(self.db_path.parent, self._db_wrapper, selected_network)
+            height_map = await BlockHeightMap.create(self.db_path.parent, self.db_wrapper, selected_network)
+            consensus_store = await ConsensusStoreSQLite3.create(self.block_store, self.coin_store, height_map)
             self._blockchain = await Blockchain.create(
-                coin_store=self.coin_store,
-                block_store=self.block_store,
+                consensus_store=consensus_store,
                 consensus_constants=self.constants,
-                height_map=height_map,
                 reserved_cores=reserved_cores,
                 single_threaded=single_threaded,
                 log_coins=log_coins,
@@ -419,7 +418,7 @@ def blockchain(self) -> Blockchain:
         return self._blockchain
 
     @property
-    def coin_store(self) -> CoinStoreProtocol:
+    def coin_store(self) -> CoinStore:
         assert self._coin_store is not None
         return self._coin_store
 
diff --git a/chia/full_node/full_node_rpc_api.py b/chia/full_node/full_node_rpc_api.py
index 3330583ee67a..875e2917db06 100644
--- a/chia/full_node/full_node_rpc_api.py
+++ b/chia/full_node/full_node_rpc_api.py
@@ -470,7 +470,7 @@ async def get_block_records(self, request: dict[str, Any]) -> EndpointResult:
             record: Optional[BlockRecord] = self.service.blockchain.try_block_record(header_hash)
             if record is None:
                 # Fetch from DB
-                record = await self.service.blockchain.block_store.get_block_record(header_hash)
+                record = await self.service.block_store.get_block_record(header_hash)
             if record is None:
                 raise ValueError(f"Block {header_hash.hex()} does not exist")
 
@@ -538,7 +538,7 @@ async def get_block_record_by_height(self, request: dict[str, Any]) -> EndpointR
         record: Optional[BlockRecord] = self.service.blockchain.try_block_record(header_hash)
         if record is None:
             # Fetch from DB
-            record = await self.service.blockchain.block_store.get_block_record(header_hash)
+            record = await self.service.block_store.get_block_record(header_hash)
         if record is None:
             raise ValueError(f"Block {header_hash} does not exist")
         return {"block_record": record}
@@ -551,7 +551,7 @@ async def get_block_record(self, request: dict[str, Any]) -> EndpointResult:
         record: Optional[BlockRecord] = self.service.blockchain.try_block_record(header_hash)
         if record is None:
             # Fetch from DB
-            record = await self.service.blockchain.block_store.get_block_record(header_hash)
+            record = await self.service.block_store.get_block_record(header_hash)
         if record is None:
             raise ValueError(f"Block {header_hash.hex()} does not exist")
 
@@ -634,7 +634,7 @@ async def get_coin_records_by_puzzle_hash(self, request: dict[str, Any]) -> Endp
         if "include_spent_coins" in request:
             kwargs["include_spent_coins"] = request["include_spent_coins"]
 
-        coin_records = await self.service.blockchain.coin_store.get_coin_records_by_puzzle_hash(**kwargs)
+        coin_records = await self.service.coin_store.get_coin_records_by_puzzle_hash(**kwargs)
 
         return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
 
@@ -656,7 +656,7 @@ async def get_coin_records_by_puzzle_hashes(self, request: dict[str, Any]) -> En
         if "include_spent_coins" in request:
             kwargs["include_spent_coins"] = request["include_spent_coins"]
 
-        coin_records = await self.service.blockchain.coin_store.get_coin_records_by_puzzle_hashes(**kwargs)
+        coin_records = await self.service.coin_store.get_coin_records_by_puzzle_hashes(**kwargs)
 
         return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
 
@@ -668,7 +668,7 @@ async def get_coin_record_by_name(self, request: dict[str, Any]) -> EndpointResu
             raise ValueError("Name not in request")
         name = bytes32.from_hexstr(request["name"])
 
-        coin_record: Optional[CoinRecord] = await self.service.blockchain.coin_store.get_coin_record(name)
+        coin_record: Optional[CoinRecord] = await self.service.coin_store.get_coin_record(name)
         if coin_record is None:
             raise ValueError(f"Coin record 0x{name.hex()} not found")
 
@@ -692,7 +692,7 @@ async def get_coin_records_by_names(self, request: dict[str, Any]) -> EndpointRe
         if "include_spent_coins" in request:
             kwargs["include_spent_coins"] = request["include_spent_coins"]
 
-        coin_records = await self.service.blockchain.coin_store.get_coin_records_by_names(**kwargs)
+        coin_records = await self.service.coin_store.get_coin_records_by_names(**kwargs)
 
         return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
 
@@ -714,7 +714,7 @@ async def get_coin_records_by_parent_ids(self, request: dict[str, Any]) -> Endpo
         if "include_spent_coins" in request:
             kwargs["include_spent_coins"] = request["include_spent_coins"]
 
-        coin_records = await self.service.blockchain.coin_store.get_coin_records_by_parent_ids(**kwargs)
+        coin_records = await self.service.coin_store.get_coin_records_by_parent_ids(**kwargs)
 
         return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
 
@@ -743,7 +743,7 @@ async def get_coin_records_by_hint(self, request: dict[str, Any]) -> EndpointRes
         if "include_spent_coins" in request:
             kwargs["include_spent_coins"] = request["include_spent_coins"]
 
-        coin_records = await self.service.blockchain.coin_store.get_coin_records_by_names(**kwargs)
+        coin_records = await self.service.coin_store.get_coin_records_by_names(**kwargs)
 
         return {"coin_records": [coin_record_dict_backwards_compat(cr.to_json_dict()) for cr in coin_records]}
 
@@ -1018,7 +1018,7 @@ async def get_fee_estimate(self, request: dict[str, Any]) -> dict[str, Any]:
             assert last_peak_timestamp is not None  # mypy
             assert last_tx_block.fees is not None  # mypy
 
-            record = await self.service.blockchain.block_store.get_full_block(last_tx_block.header_hash)
+            record = await self.service.block_store.get_full_block(last_tx_block.header_hash)
 
             last_block_cost = 0
             fee_rate_last_block = 0.0
diff --git a/tach.toml b/tach.toml
index e05edea91695..c46041465137 100644
--- a/tach.toml
+++ b/tach.toml
@@ -22,7 +22,6 @@ path = "chia.consensus"
 depends_on = [
     "chia.types",
     "chia.util",
-    { path = "chia.full_node", deprecated = false },
 ]
 
 [[modules]]