Merge branch 'long_lived/datalayer_merkle_blob' into add_rs_merkle_blob

chia/_tests/core/data_layer/test_data_rpc.py

@@ -46,12 +46,13 @@ from chia.data_layer.data_layer_util import (
     ProofLayer,
     Status,
     StoreProofs,
+    get_delta_filename_path,
+    get_full_tree_filename_path,
     key_hash,
     leaf_hash,
 )
 from chia.data_layer.data_layer_wallet import DataLayerWallet, verify_offer
 from chia.data_layer.data_store import DataStore
-from chia.data_layer.download_data import get_delta_filename_path, get_full_tree_filename_path
 from chia.rpc.data_layer_rpc_api import DataLayerRpcApi
 from chia.rpc.data_layer_rpc_client import DataLayerRpcClient
 from chia.rpc.wallet_rpc_api import WalletRpcApi
@@ -129,6 +130,7 @@ async def init_data_layer(
     manage_data_interval: int = 5,
     maximum_full_file_count: Optional[int] = None,
     group_files_by_store: bool = False,
+    enable_batch_autoinsert: bool = True,
 ) -> AsyncIterator[DataLayer]:
     async with init_data_layer_service(
         wallet_rpc_port,
@@ -137,7 +139,7 @@ async def init_data_layer(
         wallet_service,
         manage_data_interval,
         maximum_full_file_count,
-        True,
+        enable_batch_autoinsert,
         group_files_by_store,
     ) as data_layer_service:
         yield data_layer_service._api.data_layer
@@ -251,6 +253,7 @@ def create_mnemonic(seed: bytes = b"ab") -> str:
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_create_insert_get(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -330,6 +333,7 @@ async def test_create_insert_get(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_upsert(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -360,6 +364,7 @@ async def test_upsert(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_create_double_insert(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -397,6 +402,7 @@ async def test_create_double_insert(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_keys_values_ancestors(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -443,7 +449,7 @@ async def test_keys_values_ancestors(
             assert key in dic
         val = await data_rpc_api.get_ancestors({"id": store_id.hex(), "hash": val["keys_values"][4]["hash"]})
         # todo better assertions for get_ancestors result
-        assert len(val["ancestors"]) == 3
+        assert len(val["ancestors"]) == 1
         res_before = await data_rpc_api.get_root({"id": store_id.hex()})
         assert res_before["confirmed"] is True
         assert res_before["timestamp"] > 0
@@ -473,6 +479,7 @@ async def test_keys_values_ancestors(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_get_roots(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -526,6 +533,7 @@ async def test_get_roots(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_get_root_history(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -580,6 +588,7 @@ async def test_get_root_history(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_get_kv_diff(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -647,13 +656,19 @@ async def test_get_kv_diff(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_batch_update_matches_single_operations(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
     wallet_rpc_api, full_node_api, wallet_rpc_port, ph, bt = await init_wallet_and_node(
         self_hostname, one_wallet_and_one_simulator_services
     )
-    async with init_data_layer(wallet_rpc_port=wallet_rpc_port, bt=bt, db_path=tmp_path) as data_layer:
+    async with init_data_layer(
+        wallet_rpc_port=wallet_rpc_port,
+        bt=bt,
+        db_path=tmp_path,
+        enable_batch_autoinsert=False,
+    ) as data_layer:
         data_rpc_api = DataLayerRpcApi(data_layer)
         res = await data_rpc_api.create_data_store({})
         assert res is not None
@@ -719,6 +734,7 @@ async def test_batch_update_matches_single_operations(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_get_owned_stores(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -759,6 +775,7 @@ async def test_get_owned_stores(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_subscriptions(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -1591,6 +1608,7 @@ make_one_take_one_unpopulated_reference = MakeAndTakeReference(
     indirect=["offer_setup"],
 )
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_make_and_take_offer(offer_setup: OfferSetup, reference: MakeAndTakeReference) -> None:
     offer_setup = await populate_offer_setup(offer_setup=offer_setup, count=reference.entries_to_insert)
 
@@ -1703,6 +1721,7 @@ async def test_make_and_then_take_offer_invalid_inclusion_key(
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_verify_offer_rpc_valid(bare_data_layer_api: DataLayerRpcApi) -> None:
     reference = make_one_take_one_reference
 
@@ -1721,6 +1740,7 @@ async def test_verify_offer_rpc_valid(bare_data_layer_api: DataLayerRpcApi) -> N
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_verify_offer_rpc_invalid(bare_data_layer_api: DataLayerRpcApi) -> None:
     reference = make_one_take_one_reference
     broken_taker_offer = copy.deepcopy(reference.make_offer_response)
@@ -1741,6 +1761,7 @@ async def test_verify_offer_rpc_invalid(bare_data_layer_api: DataLayerRpcApi) ->
 
 
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_make_offer_failure_rolls_back_db(offer_setup: OfferSetup) -> None:
     # TODO: only needs the maker and db?  wallet?
     reference = make_one_take_one_reference
@@ -1783,6 +1804,7 @@ async def test_make_offer_failure_rolls_back_db(offer_setup: OfferSetup) -> None
     ],
 )
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_make_and_cancel_offer(offer_setup: OfferSetup, reference: MakeAndTakeReference) -> None:
     offer_setup = await populate_offer_setup(offer_setup=offer_setup, count=reference.entries_to_insert)
 
@@ -1859,6 +1881,7 @@ async def test_make_and_cancel_offer(offer_setup: OfferSetup, reference: MakeAnd
     ],
 )
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_make_and_cancel_offer_then_update(
     offer_setup: OfferSetup, reference: MakeAndTakeReference, secure: bool
 ) -> None:
@@ -1948,6 +1971,7 @@ async def test_make_and_cancel_offer_then_update(
     ],
 )
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_make_and_cancel_offer_not_secure_clears_pending_roots(
     offer_setup: OfferSetup,
     reference: MakeAndTakeReference,
@@ -1990,6 +2014,7 @@ async def test_make_and_cancel_offer_not_secure_clears_pending_roots(
 
 @pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 @pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 async def test_get_sync_status(
     self_hostname: str, one_wallet_and_one_simulator_services: SimulatorsAndWalletsServices, tmp_path: Path
 ) -> None:
@@ -3020,7 +3045,12 @@ async def test_pagination_cmds(
     wallet_rpc_api, full_node_api, wallet_rpc_port, ph, bt = await init_wallet_and_node(
         self_hostname, one_wallet_and_one_simulator_services
     )
-    async with init_data_layer_service(wallet_rpc_port=wallet_rpc_port, bt=bt, db_path=tmp_path) as data_layer_service:
+    async with init_data_layer_service(
+        wallet_rpc_port=wallet_rpc_port,
+        bt=bt,
+        db_path=tmp_path,
+        enable_batch_autoinsert=False,
+    ) as data_layer_service:
         assert data_layer_service.rpc_server is not None
         rpc_port = data_layer_service.rpc_server.listen_port
         data_layer = data_layer_service._api.data_layer
@@ -3168,7 +3198,7 @@ async def test_pagination_cmds(
         if max_page_size is None or max_page_size == 100:
             assert keys == {
                 "keys": ["0x61616161", "0x6161"],
-                "root_hash": "0x889a4a61b17be799ae9d36831246672ef857a24091f54481431a83309d4e890e",
+                "root_hash": "0x3f4ae7b8e10ef48b3114843537d5def989ee0a3b6568af7e720a71730f260fa1",
                 "success": True,
                 "total_bytes": 6,
                 "total_pages": 1,
@@ -3188,7 +3218,7 @@ async def test_pagination_cmds(
                         "value": "0x6161",
                     },
                 ],
-                "root_hash": "0x889a4a61b17be799ae9d36831246672ef857a24091f54481431a83309d4e890e",
+                "root_hash": "0x3f4ae7b8e10ef48b3114843537d5def989ee0a3b6568af7e720a71730f260fa1",
                 "success": True,
                 "total_bytes": 9,
                 "total_pages": 1,
@@ -3205,7 +3235,7 @@ async def test_pagination_cmds(
         elif max_page_size == 5:
             assert keys == {
                 "keys": ["0x61616161"],
-                "root_hash": "0x889a4a61b17be799ae9d36831246672ef857a24091f54481431a83309d4e890e",
+                "root_hash": "0x3f4ae7b8e10ef48b3114843537d5def989ee0a3b6568af7e720a71730f260fa1",
                 "success": True,
                 "total_bytes": 6,
                 "total_pages": 2,
@@ -3219,7 +3249,7 @@ async def test_pagination_cmds(
                         "value": "0x61",
                     }
                 ],
-                "root_hash": "0x889a4a61b17be799ae9d36831246672ef857a24091f54481431a83309d4e890e",
+                "root_hash": "0x3f4ae7b8e10ef48b3114843537d5def989ee0a3b6568af7e720a71730f260fa1",
                 "success": True,
                 "total_bytes": 9,
                 "total_pages": 2,
@@ -3669,6 +3699,7 @@ async def test_multistore_update(
             await data_rpc_api.multistore_batch_update({"store_updates": store_updates})
 
 
+@pytest.mark.skip
 @pytest.mark.limit_consensus_modes(reason="does not depend on consensus rules")
 @pytest.mark.anyio
 async def test_unsubmitted_batch_db_migration(

chia/_tests/core/data_layer/test_data_store_schema.py

@@ -1,171 +1,17 @@
 from __future__ import annotations
 
 import sqlite3
-from typing import Any
 
 import pytest
 
-from chia._tests.core.data_layer.util import add_01234567_example, create_valid_node_values
-from chia.data_layer.data_layer_util import NodeType, Side, Status
+from chia._tests.core.data_layer.util import add_01234567_example
+from chia.data_layer.data_layer_util import Status
 from chia.data_layer.data_store import DataStore
 from chia.types.blockchain_format.sized_bytes import bytes32
 
 pytestmark = pytest.mark.data_layer
 
 
-@pytest.mark.anyio
-async def test_node_update_fails(data_store: DataStore, store_id: bytes32) -> None:
-    await add_01234567_example(data_store=data_store, store_id=store_id)
-    node = await data_store.get_node_by_key(key=b"\x04", store_id=store_id)
-
-    async with data_store.db_wrapper.writer() as writer:
-        with pytest.raises(sqlite3.IntegrityError, match=r"^updates not allowed to the node table$"):
-            await writer.execute(
-                "UPDATE node SET value = :value WHERE hash == :hash",
-                {
-                    "hash": node.hash,
-                    "value": node.value,
-                },
-            )
-
-
-@pytest.mark.parametrize(argnames="length", argvalues=sorted(set(range(50)) - {32}))
-@pytest.mark.anyio
-async def test_node_hash_must_be_32(
-    data_store: DataStore,
-    store_id: bytes32,
-    length: int,
-    valid_node_values: dict[str, Any],
-) -> None:
-    valid_node_values["hash"] = bytes([0] * length)
-
-    async with data_store.db_wrapper.writer() as writer:
-        with pytest.raises(sqlite3.IntegrityError, match=r"^CHECK constraint failed:"):
-            await writer.execute(
-                """
-                INSERT INTO node(hash, node_type, left, right, key, value)
-                VALUES(:hash, :node_type, :left, :right, :key, :value)
-                """,
-                valid_node_values,
-            )
-
-
-@pytest.mark.anyio
-async def test_node_hash_must_not_be_null(
-    data_store: DataStore,
-    store_id: bytes32,
-    valid_node_values: dict[str, Any],
-) -> None:
-    valid_node_values["hash"] = None
-
-    async with data_store.db_wrapper.writer() as writer:
-        with pytest.raises(sqlite3.IntegrityError, match=r"^NOT NULL constraint failed: node.hash$"):
-            await writer.execute(
-                """
-                INSERT INTO node(hash, node_type, left, right, key, value)
-                VALUES(:hash, :node_type, :left, :right, :key, :value)
-                """,
-                valid_node_values,
-            )
-
-
-@pytest.mark.anyio
-async def test_node_type_must_be_valid(
-    data_store: DataStore,
-    node_type: NodeType,
-    bad_node_type: int,
-    valid_node_values: dict[str, Any],
-) -> None:
-    valid_node_values["node_type"] = bad_node_type
-
-    async with data_store.db_wrapper.writer() as writer:
-        with pytest.raises(sqlite3.IntegrityError, match=r"^CHECK constraint failed:"):
-            await writer.execute(
-                """
-                INSERT INTO node(hash, node_type, left, right, key, value)
-                VALUES(:hash, :node_type, :left, :right, :key, :value)
-                """,
-                valid_node_values,
-            )
-
-
-@pytest.mark.parametrize(argnames="side", argvalues=Side)
-@pytest.mark.anyio
-async def test_node_internal_child_not_null(data_store: DataStore, store_id: bytes32, side: Side) -> None:
-    await add_01234567_example(data_store=data_store, store_id=store_id)
-    node_a = await data_store.get_node_by_key(key=b"\x02", store_id=store_id)
-    node_b = await data_store.get_node_by_key(key=b"\x04", store_id=store_id)
-
-    values = create_valid_node_values(node_type=NodeType.INTERNAL, left_hash=node_a.hash, right_hash=node_b.hash)
-
-    if side == Side.LEFT:
-        values["left"] = None
-    elif side == Side.RIGHT:
-        values["right"] = None
-
-    async with data_store.db_wrapper.writer() as writer:
-        with pytest.raises(sqlite3.IntegrityError, match=r"^CHECK constraint failed:"):
-            await writer.execute(
-                """
-                INSERT INTO node(hash, node_type, left, right, key, value)
-                VALUES(:hash, :node_type, :left, :right, :key, :value)
-                """,
-                values,
-            )
-
-
-@pytest.mark.parametrize(argnames="bad_child_hash", argvalues=[b"\x01" * 32, b"\0" * 31, b""])
-@pytest.mark.parametrize(argnames="side", argvalues=Side)
-@pytest.mark.anyio
-async def test_node_internal_must_be_valid_reference(
-    data_store: DataStore,
-    store_id: bytes32,
-    bad_child_hash: bytes,
-    side: Side,
-) -> None:
-    await add_01234567_example(data_store=data_store, store_id=store_id)
-    node_a = await data_store.get_node_by_key(key=b"\x02", store_id=store_id)
-    node_b = await data_store.get_node_by_key(key=b"\x04", store_id=store_id)
-
-    values = create_valid_node_values(node_type=NodeType.INTERNAL, left_hash=node_a.hash, right_hash=node_b.hash)
-
-    if side == Side.LEFT:
-        values["left"] = bad_child_hash
-    elif side == Side.RIGHT:
-        values["right"] = bad_child_hash
-    else:  # pragma: no cover
-        assert False
-
-    async with data_store.db_wrapper.writer() as writer:
-        with pytest.raises(sqlite3.IntegrityError, match=r"^FOREIGN KEY constraint failed$"):
-            await writer.execute(
-                """
-                INSERT INTO node(hash, node_type, left, right, key, value)
-                VALUES(:hash, :node_type, :left, :right, :key, :value)
-                """,
-                values,
-            )
-
-
-@pytest.mark.parametrize(argnames="key_or_value", argvalues=["key", "value"])
-@pytest.mark.anyio
-async def test_node_terminal_key_value_not_null(data_store: DataStore, store_id: bytes32, key_or_value: str) -> None:
-    await add_01234567_example(data_store=data_store, store_id=store_id)
-
-    values = create_valid_node_values(node_type=NodeType.TERMINAL)
-    values[key_or_value] = None
-
-    async with data_store.db_wrapper.writer() as writer:
-        with pytest.raises(sqlite3.IntegrityError, match=r"^CHECK constraint failed:"):
-            await writer.execute(
-                """
-                INSERT INTO node(hash, node_type, left, right, key, value)
-                VALUES(:hash, :node_type, :left, :right, :key, :value)
-                """,
-                values,
-            )
-
-
 @pytest.mark.parametrize(argnames="length", argvalues=sorted(set(range(50)) - {32}))
 @pytest.mark.anyio
 async def test_root_store_id_must_be_32(data_store: DataStore, store_id: bytes32, length: int) -> None:
@@ -250,21 +96,6 @@ async def test_root_generation_must_not_be_null(data_store: DataStore, store_id:
             )
 
 
-@pytest.mark.anyio
-async def test_root_node_hash_must_reference(data_store: DataStore) -> None:
-    values = {"tree_id": bytes32.zeros, "generation": 0, "node_hash": bytes32.zeros, "status": Status.PENDING}
-
-    async with data_store.db_wrapper.writer() as writer:
-        with pytest.raises(sqlite3.IntegrityError, match=r"^FOREIGN KEY constraint failed$"):
-            await writer.execute(
-                """
-                INSERT INTO root(tree_id, generation, node_hash, status)
-                VALUES(:tree_id, :generation, :node_hash, :status)
-                """,
-                values,
-            )
-
-
 @pytest.mark.parametrize(argnames="bad_status", argvalues=sorted(set(range(-20, 20)) - {*Status}))
 @pytest.mark.anyio
 async def test_root_status_must_be_valid(data_store: DataStore, store_id: bytes32, bad_status: int) -> None:
@@ -319,44 +150,6 @@ async def test_root_store_id_generation_must_be_unique(data_store: DataStore, st
             )
 
 
-@pytest.mark.parametrize(argnames="length", argvalues=sorted(set(range(50)) - {32}))
-@pytest.mark.anyio
-async def test_ancestors_ancestor_must_be_32(
-    data_store: DataStore,
-    store_id: bytes32,
-    length: int,
-) -> None:
-    async with data_store.db_wrapper.writer() as writer:
-        node_hash = await data_store._insert_terminal_node(key=b"\x00", value=b"\x01")
-        with pytest.raises(sqlite3.IntegrityError, match=r"^CHECK constraint failed:"):
-            await writer.execute(
-                """
-                INSERT INTO ancestors(hash, ancestor, tree_id, generation)
-                VALUES(:hash, :ancestor, :tree_id, :generation)
-                """,
-                {"hash": node_hash, "ancestor": bytes([0] * length), "tree_id": bytes32.zeros, "generation": 0},
-            )
-
-
-@pytest.mark.parametrize(argnames="length", argvalues=sorted(set(range(50)) - {32}))
-@pytest.mark.anyio
-async def test_ancestors_store_id_must_be_32(
-    data_store: DataStore,
-    store_id: bytes32,
-    length: int,
-) -> None:
-    async with data_store.db_wrapper.writer() as writer:
-        node_hash = await data_store._insert_terminal_node(key=b"\x00", value=b"\x01")
-        with pytest.raises(sqlite3.IntegrityError, match=r"^CHECK constraint failed:"):
-            await writer.execute(
-                """
-                INSERT INTO ancestors(hash, ancestor, tree_id, generation)
-                VALUES(:hash, :ancestor, :tree_id, :generation)
-                """,
-                {"hash": node_hash, "ancestor": bytes32.zeros, "tree_id": bytes([0] * length), "generation": 0},
-            )
-
-
 @pytest.mark.parametrize(argnames="length", argvalues=sorted(set(range(50)) - {32}))
 @pytest.mark.anyio
 async def test_subscriptions_store_id_must_be_32(

chia/_tests/core/data_layer/test_merkle_blob.py

@@ -385,6 +385,7 @@ def test_get_raw_node_raises_for_invalid_indexes(index: TreeIndex) -> None:
 
     with pytest.raises(InvalidIndexError):
         merkle_blob.get_raw_node(index)
+
     with pytest.raises(InvalidIndexError):
         merkle_blob.get_metadata(index)
 

chia/data_layer/data_layer.py

@@ -41,18 +41,14 @@ from chia.data_layer.data_layer_util import (
     TerminalNode,
     Unspecified,
     UnsubscribeData,
+    get_delta_filename_path,
+    get_full_tree_filename_path,
     leaf_hash,
     unspecified,
 )
 from chia.data_layer.data_layer_wallet import DataLayerWallet, Mirror, SingletonRecord, verify_offer
 from chia.data_layer.data_store import DataStore
-from chia.data_layer.download_data import (
-    delete_full_file_if_exists,
-    get_delta_filename_path,
-    get_full_tree_filename_path,
-    insert_from_delta_file,
-    write_files_for_root,
-)
+from chia.data_layer.download_data import delete_full_file_if_exists, insert_from_delta_file, write_files_for_root
 from chia.rpc.rpc_server import StateChangedProtocol, default_get_connections
 from chia.rpc.wallet_request_types import LogIn
 from chia.rpc.wallet_rpc_client import WalletRpcClient
@@ -199,7 +195,7 @@ class DataLayer:
         async with DataStore.managed(database=self.db_path, sql_log_path=sql_log_path) as self._data_store:
             self._wallet_rpc = await self.wallet_rpc_init
 
-            await self._data_store.migrate_db()
+            await self._data_store.migrate_db(self.server_files_location)
             self.periodically_manage_data_task = asyncio.create_task(self.periodically_manage_data())
             try:
                 yield
@@ -254,7 +250,6 @@ class DataLayer:
     ) -> Optional[TransactionRecord]:
         status = Status.PENDING if submit_on_chain else Status.PENDING_BATCH
         await self.batch_insert(store_id=store_id, changelist=changelist, status=status)
-        await self.data_store.clean_node_table()
 
         if submit_on_chain:
             return await self.publish_update(store_id=store_id, fee=fee)
@@ -288,8 +283,6 @@ class DataLayer:
             status = Status.PENDING if submit_on_chain else Status.PENDING_BATCH
             await self.batch_insert(store_id=store_id, changelist=changelist, status=status)
 
-        await self.data_store.clean_node_table()
-
         if submit_on_chain:
             update_dictionary: dict[bytes32, bytes32] = {}
             for store_id in store_ids:
@@ -532,7 +525,6 @@ class DataLayer:
                     and pending_root.status == Status.PENDING
                 ):
                     await self.data_store.change_root_status(pending_root, Status.COMMITTED)
-                    await self.data_store.build_ancestor_table_for_latest_root(store_id=store_id)
             await self.data_store.clear_pending_roots(store_id=store_id)
 
     async def fetch_and_validate(self, store_id: bytes32) -> None:
@@ -835,8 +827,6 @@ class DataLayer:
         # stop tracking first, then unsubscribe from the data store
         await self.wallet_rpc.dl_stop_tracking(store_id)
         await self.data_store.unsubscribe(store_id)
-        if not retain_data:
-            await self.data_store.delete_store_data(store_id)
 
         self.log.info(f"Unsubscribed to {store_id}")
         for file_path in paths:
@@ -1136,7 +1126,6 @@ class DataLayer:
 
             verify_offer(maker=offer.maker, taker=offer.taker, summary=summary)
 
-        await self.data_store.clean_node_table()
         return offer
 
     async def take_offer(
@@ -1195,8 +1184,6 @@ class DataLayer:
                 },
             }
 
-        await self.data_store.clean_node_table()
-
         # Excluding wallet from transaction since failures in the wallet may occur
         # after the transaction is submitted to the chain.  If we roll back data we
         # may lose published data.

chia/data_layer/data_layer_errors.py

@@ -38,6 +38,11 @@ class KeyNotFoundError(Exception):
         super().__init__(f"Key not found: {key.hex()}")
 
 
+class MerkleBlobNotFoundError(Exception):
+    def __init__(self, root_hash: bytes32) -> None:
+        super().__init__(f"Cannot find merkle blob for root hash {root_hash.hex()}")
+
+
 class OfferIntegrityError(Exception):
     pass
 

chia/data_layer/data_layer_util.py

@@ -4,6 +4,7 @@ import dataclasses
 from dataclasses import dataclass, field
 from enum import Enum, IntEnum
 from hashlib import sha256
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional, Union
 
 import aiosqlite
@@ -48,6 +49,44 @@ def key_hash(key: bytes) -> bytes32:
     return bytes32(sha256(b"\1" + key).digest())
 
 
+def get_full_tree_filename(store_id: bytes32, node_hash: bytes32, generation: int, group_by_store: bool = False) -> str:
+    if group_by_store:
+        return f"{store_id}/{node_hash}-full-{generation}-v1.0.dat"
+    return f"{store_id}-{node_hash}-full-{generation}-v1.0.dat"
+
+
+def get_delta_filename(store_id: bytes32, node_hash: bytes32, generation: int, group_by_store: bool = False) -> str:
+    if group_by_store:
+        return f"{store_id}/{node_hash}-delta-{generation}-v1.0.dat"
+    return f"{store_id}-{node_hash}-delta-{generation}-v1.0.dat"
+
+
+def get_full_tree_filename_path(
+    foldername: Path,
+    store_id: bytes32,
+    node_hash: bytes32,
+    generation: int,
+    group_by_store: bool = False,
+) -> Path:
+    if group_by_store:
+        path = foldername.joinpath(f"{store_id}")
+        return path.joinpath(f"{node_hash}-full-{generation}-v1.0.dat")
+    return foldername.joinpath(f"{store_id}-{node_hash}-full-{generation}-v1.0.dat")
+
+
+def get_delta_filename_path(
+    foldername: Path,
+    store_id: bytes32,
+    node_hash: bytes32,
+    generation: int,
+    group_by_store: bool = False,
+) -> Path:
+    if group_by_store:
+        path = foldername.joinpath(f"{store_id}")
+        return path.joinpath(f"{node_hash}-delta-{generation}-v1.0.dat")
+    return foldername.joinpath(f"{store_id}-{node_hash}-delta-{generation}-v1.0.dat")
+
+
 @dataclasses.dataclass(frozen=True)
 class PaginationData:
     total_pages: int

chia/data_layer/download_data.py

@@ -10,49 +10,19 @@ from typing import Optional
 import aiohttp
 from typing_extensions import Literal
 
-from chia.data_layer.data_layer_util import NodeType, PluginRemote, Root, SerializedNode, ServerInfo, Status
+from chia.data_layer.data_layer_util import (
+    PluginRemote,
+    Root,
+    ServerInfo,
+    get_delta_filename,
+    get_delta_filename_path,
+    get_full_tree_filename,
+    get_full_tree_filename_path,
+)
 from chia.data_layer.data_store import DataStore
 from chia.types.blockchain_format.sized_bytes import bytes32
 
 
-def get_full_tree_filename(store_id: bytes32, node_hash: bytes32, generation: int, group_by_store: bool = False) -> str:
-    if group_by_store:
-        return f"{store_id}/{node_hash}-full-{generation}-v1.0.dat"
-    return f"{store_id}-{node_hash}-full-{generation}-v1.0.dat"
-
-
-def get_delta_filename(store_id: bytes32, node_hash: bytes32, generation: int, group_by_store: bool = False) -> str:
-    if group_by_store:
-        return f"{store_id}/{node_hash}-delta-{generation}-v1.0.dat"
-    return f"{store_id}-{node_hash}-delta-{generation}-v1.0.dat"
-
-
-def get_full_tree_filename_path(
-    foldername: Path,
-    store_id: bytes32,
-    node_hash: bytes32,
-    generation: int,
-    group_by_store: bool = False,
-) -> Path:
-    if group_by_store:
-        path = foldername.joinpath(f"{store_id}")
-        return path.joinpath(f"{node_hash}-full-{generation}-v1.0.dat")
-    return foldername.joinpath(f"{store_id}-{node_hash}-full-{generation}-v1.0.dat")
-
-
-def get_delta_filename_path(
-    foldername: Path,
-    store_id: bytes32,
-    node_hash: bytes32,
-    generation: int,
-    group_by_store: bool = False,
-) -> Path:
-    if group_by_store:
-        path = foldername.joinpath(f"{store_id}")
-        return path.joinpath(f"{node_hash}-delta-{generation}-v1.0.dat")
-    return foldername.joinpath(f"{store_id}-{node_hash}-delta-{generation}-v1.0.dat")
-
-
 def is_filename_valid(filename: str, group_by_store: bool = False) -> bool:
     if group_by_store:
         if filename.count("/") != 1:
@@ -87,45 +57,6 @@ def is_filename_valid(filename: str, group_by_store: bool = False) -> bool:
     return reformatted == filename
 
 
-async def insert_into_data_store_from_file(
-    data_store: DataStore,
-    store_id: bytes32,
-    root_hash: Optional[bytes32],
-    filename: Path,
-) -> int:
-    num_inserted = 0
-    with open(filename, "rb") as reader:
-        while True:
-            chunk = b""
-            while len(chunk) < 4:
-                size_to_read = 4 - len(chunk)
-                cur_chunk = reader.read(size_to_read)
-                if cur_chunk is None or cur_chunk == b"":
-                    if size_to_read < 4:
-                        raise Exception("Incomplete read of length.")
-                    break
-                chunk += cur_chunk
-            if chunk == b"":
-                break
-
-            size = int.from_bytes(chunk, byteorder="big")
-            serialize_nodes_bytes = b""
-            while len(serialize_nodes_bytes) < size:
-                size_to_read = size - len(serialize_nodes_bytes)
-                cur_chunk = reader.read(size_to_read)
-                if cur_chunk is None or cur_chunk == b"":
-                    raise Exception("Incomplete read of blob.")
-                serialize_nodes_bytes += cur_chunk
-            serialized_node = SerializedNode.from_bytes(serialize_nodes_bytes)
-
-            node_type = NodeType.TERMINAL if serialized_node.is_terminal else NodeType.INTERNAL
-            await data_store.insert_node(node_type, serialized_node.value1, serialized_node.value2)
-            num_inserted += 1
-
-    await data_store.insert_root_with_ancestor_table(store_id=store_id, node_hash=root_hash, status=Status.COMMITTED)
-    return num_inserted
-
-
 @dataclass
 class WriteFilesResult:
     result: bool
@@ -288,15 +219,14 @@ async def insert_from_delta_file(
                 existing_generation,
                 group_files_by_store,
             )
-            num_inserted = await insert_into_data_store_from_file(
-                data_store,
+            await data_store.insert_into_data_store_from_file(
                 store_id,
                 None if root_hash == bytes32.zeros else root_hash,
                 target_filename_path,
             )
             log.info(
                 f"Successfully inserted hash {root_hash} from delta file. "
-                f"Generation: {existing_generation}. Store id: {store_id}. Nodes inserted: {num_inserted}."
+                f"Generation: {existing_generation}. Store id: {store_id}."
             )
 
             if target_generation - existing_generation <= maximum_full_file_count - 1:
@@ -386,4 +316,4 @@ async def http_download(
                     new_percentage = f"{progress_byte / size:.0%}"
                     if new_percentage != progress_percentage:
                         progress_percentage = new_percentage
-                        log.debug(f"Downloading delta file {filename}. {progress_percentage} of {size} bytes.")
+                        log.info(f"Downloading delta file {filename}. {progress_percentage} of {size} bytes.")

chia/data_layer/util/benchmark.py

@@ -6,14 +6,13 @@ import sys
 import tempfile
 import time
 from pathlib import Path
-from typing import Optional
 
-from chia.data_layer.data_layer_util import Side, TerminalNode, leaf_hash
+from chia.data_layer.data_layer_util import Side, Status, leaf_hash
 from chia.data_layer.data_store import DataStore
 from chia.types.blockchain_format.sized_bytes import bytes32
 
 
-async def generate_datastore(num_nodes: int, slow_mode: bool) -> None:
+async def generate_datastore(num_nodes: int) -> None:
     with tempfile.TemporaryDirectory() as temp_directory:
         temp_directory_path = Path(temp_directory)
         db_path = temp_directory_path.joinpath("dl_benchmark.sqlite")
@@ -22,9 +21,10 @@ async def generate_datastore(num_nodes: int, slow_mode: bool) -> None:
         if os.path.exists(db_path):
             os.remove(db_path)
 
+        start_time = time.monotonic()
         async with DataStore.managed(database=db_path) as data_store:
             store_id = bytes32(b"0" * 32)
-            await data_store.create_tree(store_id)
+            await data_store.create_tree(store_id, status=Status.COMMITTED)
 
             insert_time = 0.0
             insert_count = 0
@@ -36,58 +36,40 @@ async def generate_datastore(num_nodes: int, slow_mode: bool) -> None:
             for i in range(num_nodes):
                 key = i.to_bytes(4, byteorder="big")
                 value = (2 * i).to_bytes(4, byteorder="big")
-                seed = leaf_hash(key=key, value=value)
-                reference_node_hash: Optional[bytes32] = await data_store.get_terminal_node_for_seed(store_id, seed)
-                side: Optional[Side] = data_store.get_side_for_seed(seed)
+                seed = leaf_hash(key, value)
+                node = await data_store.get_terminal_node_for_seed(seed, store_id)
 
-                if i == 0:
-                    reference_node_hash = None
-                    side = None
                 if i % 3 == 0:
                     t1 = time.time()
-                    if not slow_mode:
-                        await data_store.insert(
-                            key=key,
-                            value=value,
-                            store_id=store_id,
-                            reference_node_hash=reference_node_hash,
-                            side=side,
-                        )
-                    else:
-                        await data_store.insert(
-                            key=key,
-                            value=value,
-                            store_id=store_id,
-                            reference_node_hash=reference_node_hash,
-                            side=side,
-                            use_optimized=False,
-                        )
+                    await data_store.autoinsert(
+                        key=key,
+                        value=value,
+                        store_id=store_id,
+                        status=Status.COMMITTED,
+                    )
+                    t2 = time.time()
+                    autoinsert_count += 1
+                elif i % 3 == 1:
+                    assert node is not None
+                    reference_node_hash = node.hash
+                    side_seed = bytes(seed)[0]
+                    side = Side.LEFT if side_seed < 128 else Side.RIGHT
+                    t1 = time.time()
+                    await data_store.insert(
+                        key=key,
+                        value=value,
+                        store_id=store_id,
+                        reference_node_hash=reference_node_hash,
+                        side=side,
+                        status=Status.COMMITTED,
+                    )
                     t2 = time.time()
                     insert_time += t2 - t1
                     insert_count += 1
-                elif i % 3 == 1:
-                    t1 = time.time()
-                    if not slow_mode:
-                        await data_store.autoinsert(key=key, value=value, store_id=store_id)
-                    else:
-                        await data_store.autoinsert(
-                            key=key,
-                            value=value,
-                            store_id=store_id,
-                            use_optimized=False,
-                        )
-                    t2 = time.time()
-                    autoinsert_time += t2 - t1
-                    autoinsert_count += 1
                 else:
                     t1 = time.time()
-                    assert reference_node_hash is not None
-                    node = await data_store.get_node(reference_node_hash)
-                    assert isinstance(node, TerminalNode)
-                    if not slow_mode:
-                        await data_store.delete(key=node.key, store_id=store_id)
-                    else:
-                        await data_store.delete(key=node.key, store_id=store_id, use_optimized=False)
+                    assert node is not None
+                    await data_store.delete(key=node.key, store_id=store_id, status=Status.COMMITTED)
                     t2 = time.time()
                     delete_time += t2 - t1
                     delete_count += 1
@@ -95,13 +77,12 @@ async def generate_datastore(num_nodes: int, slow_mode: bool) -> None:
             print(f"Average insert time: {insert_time / insert_count}")
             print(f"Average autoinsert time: {autoinsert_time / autoinsert_count}")
             print(f"Average delete time: {delete_time / delete_count}")
-            print(f"Total time for {num_nodes} operations: {insert_time + autoinsert_time + delete_time}")
+            print(f"Total time for {num_nodes} operations: {insert_time + delete_time + autoinsert_time}")
             root = await data_store.get_tree_root(store_id=store_id)
             print(f"Root hash: {root.node_hash}")
+            finish_time = time.monotonic()
+            print(f"Total runtime: {finish_time - start_time}")
 
 
 if __name__ == "__main__":
-    slow_mode = False
-    if len(sys.argv) > 2 and sys.argv[2] == "slow":
-        slow_mode = True
-    asyncio.run(generate_datastore(int(sys.argv[1]), slow_mode))
+    asyncio.run(generate_datastore(int(sys.argv[1])))

chia/data_layer/util/merkle_blob.py

@@ -208,16 +208,16 @@ class MerkleBlob:
         self.blob[data_start:end] = pack_raw_node(new_node)
 
     def get_random_leaf_node(self, seed: bytes) -> RawLeafMerkleNode:
+        path = "".join(reversed("".join(f"{b:08b}" for b in seed)))
         node = self.get_raw_node(TreeIndex(0))
-        for byte in seed:
-            for bit in range(8):
-                if isinstance(node, RawLeafMerkleNode):
-                    return node
-                assert isinstance(node, RawInternalMerkleNode)
-                if byte & (1 << bit):
-                    node = self.get_raw_node(node.left)
-                else:
-                    node = self.get_raw_node(node.right)
+        for bit in path:
+            if isinstance(node, RawLeafMerkleNode):
+                return node
+            assert isinstance(node, RawInternalMerkleNode)
+            if bit == "0":
+                node = self.get_raw_node(node.left)
+            else:
+                node = self.get_raw_node(node.right)
 
         raise Exception("Cannot find leaf from seed")
 
@@ -239,6 +239,22 @@ class MerkleBlob:
 
         return key_to_index
 
+    def get_hashes_indexes(self) -> dict[bytes32, TreeIndex]:
+        if len(self.blob) == 0:
+            return {}
+
+        hash_to_index: dict[bytes32, TreeIndex] = {}
+        queue: list[TreeIndex] = [TreeIndex(0)]
+        while len(queue) > 0:
+            node_index = queue.pop()
+            node = self.get_raw_node(node_index)
+            hash_to_index[bytes32(node.hash)] = node_index
+            if isinstance(node, RawInternalMerkleNode):
+                queue.append(node.left)
+                queue.append(node.right)
+
+        return hash_to_index
+
     def get_keys_values(self) -> dict[KVId, KVId]:
         if len(self.blob) == 0:
             return {}
@@ -326,6 +342,9 @@ class MerkleBlob:
         if isinstance(new_node, RawLeafMerkleNode):
             self.key_to_index[new_node.key] = new_index
 
+    def key_exists(self, key: KVId) -> bool:
+        return key in self.key_to_index
+
     def insert(
         self,
         key: KVId,
@@ -359,7 +378,10 @@ class MerkleBlob:
 
         if len(self.key_to_index) == 1:
             self.blob.clear()
-            internal_node_hash = internal_hash(bytes32(old_leaf.hash), bytes32(hash))
+            if side == Side.LEFT:
+                internal_node_hash = internal_hash(bytes32(hash), bytes32(old_leaf.hash))
+            else:
+                internal_node_hash = internal_hash(bytes32(old_leaf.hash), bytes32(hash))
             self.blob.extend(
                 NodeMetadata(type=NodeType.internal, dirty=False).pack()
                 + pack_raw_node(
@@ -476,6 +498,7 @@ class MerkleBlob:
             return this
 
         assert isinstance(node, RawInternalMerkleNode)
+
         left_nodes = self.get_nodes_with_indexes(node.left)
         right_nodes = self.get_nodes_with_indexes(node.right)
 

chia/rpc/data_layer_rpc_api.py

@@ -610,7 +610,7 @@ class DataLayerRpcApi:
         for key in request.keys:
             node = await self.service.data_store.get_node_by_key(store_id=request.store_id, key=key)
             pi = await self.service.data_store.get_proof_of_inclusion_by_hash(
-                store_id=request.store_id, node_hash=node.hash, use_optimized=True
+                store_id=request.store_id, node_hash=node.hash
             )
 
             proof = HashOnlyProof.from_key_value(