starred/chia-blockchain
1
0
Fork 0
mirror of https://github.com/Chia-Network/chia-blockchain.git synced 2026-09-27 17:46:28 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'atari' into fc.dl_optimizations

Browse Source
This commit is contained in:
Florin Chirica
2022-02-23 19:25:09 +02:00
parent afae169999 925356a42b
commit 0675f6a293
25 changed files with 953 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/build-windows-installer.yml
+1 -1
View File
@@ -13,7 +13,7 @@ on:
jobs:
build:
name: Windows Installer on Windows 10 and Python 3.9
runs-on: [windows-latest]
runs-on: [windows-2019]
timeout-minutes: 40
steps:
chia/cmds/configure.py
+16
View File
@@ -19,6 +19,7 @@ def configure(
set_peer_count: str,
testnet: str,
peer_connect_timeout: str,
enable_data_server: str = "",
):
config: Dict = load_config(DEFAULT_ROOT_PATH, "config.yaml")
change_made = False
@@ -89,6 +90,13 @@ def configure(
config["full_node"]["target_peer_count"] = int(set_peer_count)
print("Target peer count updated")
change_made = True
if enable_data_server:
config["data_layer"]["run_server"] = str2bool(enable_data_server)
if str2bool(enable_data_server):
print("Data Server enabled.")
else:
print("Data Server disabled.")
change_made = True
if testnet:
if testnet == "true" or testnet == "t":
print("Setting Testnet")
@@ -197,6 +205,12 @@ def configure(
)
@click.option("--set-peer-count", help="Update the target peer count (default 80)", type=str)
@click.option("--set-peer-connect-timeout", help="Update the peer connect timeout (default 30)", type=str)
@click.option(
"--enable-data-server",
"--data-server",
help="Enable or disable data propagation server for your data layer",
type=click.Choice(["true", "t", "false", "f"]),
)
@click.pass_context
def configure_cmd(
ctx,
@@ -210,6 +224,7 @@ def configure_cmd(
set_peer_count,
testnet,
set_peer_connect_timeout,
enable_data_server,
):
configure(
ctx.obj["root_path"],
@@ -223,4 +238,5 @@ def configure_cmd(
set_peer_count,
testnet,
set_peer_connect_timeout,
enable_data_server,
)
chia/cmds/data.py
+28
View File
@@ -207,3 +207,31 @@ def unsubscribe(
from chia.cmds.data_funcs import unsubscribe_cmd
run(unsubscribe_cmd(rpc_port=data_rpc_port, store_id=id))
@data_cmd.command("get_kv_diff", short_help="")
@create_data_store_id_option()
@click.option("-hash_1", "--hash_1", help="", type=str)
@click.option("-hash_2", "--hash_2", help="", type=str)
@create_rpc_port_option()
def get_kv_diff(
id: str,
hash_1: str,
hash_2: str,
data_rpc_port: int,
) -> None:
from chia.cmds.data_funcs import get_kv_diff_cmd
run(get_kv_diff_cmd(rpc_port=data_rpc_port, store_id=id, hash_1=hash_1, hash_2=hash_2))
@data_cmd.command("get_root_history", short_help="")
@create_data_store_id_option()
@create_rpc_port_option()
def get_root_history(
id: str,
data_rpc_port: int,
) -> None:
from chia.cmds.data_funcs import get_root_history_cmd
run(get_root_history_cmd(rpc_port=data_rpc_port, store_id=id))
chia/cmds/data_funcs.py
+35
View File
@@ -154,3 +154,38 @@ async def unsubscribe_cmd(
print(f"Connection error. Check if data is running at {rpc_port}")
except Exception as e:
print(f"Exception from 'data': {e}")
async def get_kv_diff_cmd(
rpc_port: Optional[int],
store_id: str,
hash_1: str,
hash_2: str,
) -> None:
store_id_bytes = bytes32.from_hexstr(store_id)
hash_1_bytes = bytes32.from_hexstr(hash_1)
hash_2_bytes = bytes32.from_hexstr(hash_2)
try:
async with get_client(rpc_port) as (client, rpc_port):
res = await client.get_kv_diff(store_id=store_id_bytes, hash_1=hash_1_bytes, hash_2=hash_2_bytes)
print(res)
except aiohttp.ClientConnectorError:
print(f"Connection error. Check if data is running at {rpc_port}")
except Exception as e:
print(f"Exception from 'data': {e}")
async def get_root_history_cmd(
rpc_port: Optional[int],
store_id: str,
) -> None:
store_id_bytes = bytes32.from_hexstr(store_id)
try:
async with get_client(rpc_port) as (client, rpc_port):
res = await client.get_root_history(store_id=store_id_bytes)
print(res)
except aiohttp.ClientConnectorError:
print(f"Connection error. Check if data is running at {rpc_port}")
except Exception as e:
print(f"Exception from 'data': {e}")
chia/daemon/server.py
+1
View File
@@ -92,6 +92,7 @@ class PlotEvent(str, Enum):
if getattr(sys, "frozen", False):
name_map = {
"chia": "chia",
"chia_data_layer": "start_data_layer",
"chia_wallet": "start_wallet",
"chia_full_node": "start_full_node",
"chia_harvester": "start_harvester",
chia/data_layer/data_layer.py
+97 -41
View File
@@ -1,10 +1,11 @@
import logging
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple, Awaitable
from typing import Any, Callable, Dict, List, Optional, Tuple, Awaitable, Set
import aiosqlite
import traceback
import asyncio
from chia.data_layer.data_layer_types import InternalNode, TerminalNode, DownloadMode, Subscription, Root
import aiohttp
from chia.data_layer.data_layer_types import InternalNode, TerminalNode, DownloadMode, Subscription, Root, DiffData
from chia.data_layer.data_store import DataStore
from chia.rpc.wallet_rpc_client import WalletRpcClient
from chia.server.server import ChiaServer
@@ -16,10 +17,12 @@ from chia.util.path import mkdir, path_from_root
from chia.wallet.transaction_record import TransactionRecord
from chia.data_layer.data_layer_wallet import SingletonRecord
from chia.data_layer.download_data import download_data
from chia.data_layer.data_layer_server import DataLayerServer
class DataLayer:
data_store: DataStore
data_layer_server: DataLayerServer
db_wrapper: DBWrapper
db_path: Path
connection: Optional[aiosqlite.Connection]
@@ -29,6 +32,7 @@ class DataLayer:
state_changed_callback: Optional[Callable[..., object]]
wallet_id: uint64
initialized: bool
none_bytes: bytes32
def __init__(
self,
@@ -50,6 +54,8 @@ class DataLayer:
db_path_replaced: str = config["database_path"].replace("CHALLENGE", config["selected_network"])
self.db_path = path_from_root(root_path, db_path_replaced)
mkdir(self.db_path.parent)
self.data_layer_server = DataLayerServer(self.config, self.db_path, self.log)
self.none_bytes = bytes32([0] * 32)
def _set_state_changed_callback(self, callback: Callable[..., object]) -> None:
self.state_changed_callback = callback
@@ -64,16 +70,20 @@ class DataLayer:
self.wallet_rpc = await self.wallet_rpc_init
self.periodically_fetch_data_task: asyncio.Task[Any] = asyncio.create_task(self.periodically_fetch_data())
self.subscription_lock: asyncio.Lock = asyncio.Lock()
if self.config.get("run_server", False):
await self.data_layer_server.start()
return True
def _close(self) -> None:
# TODO: review for anything else we need to do here
self._shut_down = True
self.periodically_fetch_data_task.cancel()
async def _await_closed(self) -> None:
if self.connection is not None:
await self.connection.close()
if self.config.get("run_server", False):
await self.data_layer_server.stop()
self.periodically_fetch_data_task.cancel()
async def create_store(
self, fee: uint64, root: bytes32 = bytes32([0] * 32)
@@ -112,7 +122,7 @@ class DataLayer:
if root.node_hash is not None:
node_hash = root.node_hash
else:
node_hash = bytes32([0] * 32) # todo change
node_hash = self.none_bytes # todo change
transaction_record = await self.wallet_rpc.dl_update_root(tree_id, node_hash, fee)
assert transaction_record
# todo register callback to change status in data store
@@ -138,28 +148,59 @@ class DataLayer:
self.log.error("Failed to get ancestors")
return res
async def get_root(self, store_id: bytes32) -> Optional[bytes32]:
async def get_root(self, store_id: bytes32) -> Optional[SingletonRecord]:
latest = await self.wallet_rpc.dl_latest_singleton(store_id, True)
if latest is None:
self.log.error(f"Failed to get root for {store_id.hex()}")
return latest
async def get_local_root(self, store_id: bytes32) -> Optional[bytes32]:
res = await self.data_store.get_tree_root(tree_id=store_id)
if res is None:
self.log.error(f"Failed to get root for {store_id.hex()}")
return None
return res.node_hash
async def get_root_history(self, store_id: bytes32) -> List[SingletonRecord]:
records = await self.wallet_rpc.dl_history(store_id)
if records is None:
self.log.error(f"Failed to get root history for {store_id.hex()}")
root_history = []
prev: Optional[SingletonRecord] = None
for record in records:
if prev is None or record.root != prev.root:
root_history.append(record)
prev = record
return root_history
async def _validate_batch(
self,
tree_id: bytes32,
to_check: List[SingletonRecord],
min_generation: int,
max_generation: int,
) -> bool:
last_checked_hash: Optional[bytes32] = None
root: Optional[Root] = await self.data_store.get_tree_root(tree_id=tree_id)
assert root is not None
if to_check[0].root == (root.node_hash if root.node_hash is not None else self.none_bytes):
self.log.info(
f"Validated chain hash {to_check[0].root} in downloaded datastore. "
f"Wallet generation: {to_check[0].generation}"
)
else:
return False
max_generation = root.generation
last_checked_hash = to_check[0].root
to_check.pop(0)
for record in to_check:
# Ignore two consecutive identical root hashes, as we've already validated it.
if last_checked_hash is not None and record.root == last_checked_hash:
if record.root == last_checked_hash:
self.log.info(f"Skipped checking {record.root}, as it matches the previously checked hash.")
continue
# Pick the latest root in our data store with the desired hash, before our already validated data.
root: Optional[Root] = await self.data_store.get_last_tree_root_by_hash(
tree_id, record.root, max_generation
root = await self.data_store.get_last_tree_root_by_hash(
tree_id, None if record.root == self.none_bytes else record.root, max_generation
)
if root is None or root.generation < min_generation:
return False
@@ -175,7 +216,7 @@ class DataLayer:
async def fetch_and_validate(self, subscription: Subscription) -> None:
tree_id = subscription.tree_id
singleton_record: Optional[SingletonRecord] = await self.wallet_rpc.dl_latest_singleton(tree_id)
singleton_record: Optional[SingletonRecord] = await self.wallet_rpc.dl_latest_singleton(tree_id, True)
if singleton_record is None:
self.log.info(f"Fetch data: No singleton record for {tree_id}.")
return
@@ -185,6 +226,8 @@ class DataLayer:
old_root: Optional[Root] = None
try:
old_root = await self.data_store.get_tree_root(tree_id=tree_id)
except asyncio.CancelledError:
raise
except Exception:
pass
wallet_current_generation = await self.data_store.get_validated_wallet_generation(tree_id)
@@ -204,8 +247,7 @@ class DataLayer:
# TODO: wallet should handle identical hashes part?
if (
old_root is not None
and old_root.node_hash is not None
and to_check[0].root == old_root.node_hash
and to_check[0].root == (old_root.node_hash if old_root.node_hash is not None else self.none_bytes)
and len(set(record.root for record in to_check)) == 1
):
await self.data_store.set_validated_wallet_generation(tree_id, int(singleton_record.generation))
@@ -215,8 +257,8 @@ class DataLayer:
)
return
# Delete all identical root hashes to our old root hash, until we detect a change.
if old_root is not None and old_root.node_hash is not None:
while to_check[-1].root == old_root.node_hash:
if old_root is not None:
while to_check[-1].root == (old_root.node_hash if old_root.node_hash is not None else self.none_bytes):
to_check.pop()
self.log.info(
@@ -225,42 +267,51 @@ class DataLayer:
f"Target wallet generation: {singleton_record.generation}."
)
downloaded = await download_data(self.data_store, subscription, singleton_record.root)
try:
downloaded = await download_data(self.data_store, subscription, singleton_record.root)
except asyncio.CancelledError:
raise
except aiohttp.client_exceptions.ClientConnectorError:
self.log.error(f"Server unavailable for {tree_id}.")
downloaded = False
except RuntimeError as e:
self.log.error(f"Server sended invalid data for {tree_id}: {e}.")
downloaded = False
except Exception as e:
self.log.error(f"Exception while downloading data for {tree_id}: {e}.")
downloaded = False
if not downloaded:
await self.data_store.rollback_to_generation(tree_id, (0 if old_root is None else old_root.generation))
raise RuntimeError("Could not download the data.")
self.log.info(f"Successfully downloaded data for {tree_id}.")
root = await self.data_store.get_tree_root(tree_id=tree_id)
# Wallet root hash must match to our data store root hash.
if root.node_hash is not None and root.node_hash == to_check[0].root:
self.log.info(
f"Validated chain hash {root.node_hash} in downloaded datastore. "
f"Wallet generation: {to_check[0].generation}"
)
else:
raise RuntimeError("Can't find data on chain in our datastore.")
to_check.pop(0)
min_generation = (0 if old_root is None else old_root.generation) + 1
max_generation = root.generation
# Light validation: check the new set of operations against the new set of wallet records.
# If this matches, we know all data will match, as we've previously checked that data matches
# for `min_generation` data store root and `wallet_current_generation` wallet record.
is_valid: bool = await self._validate_batch(tree_id, to_check, min_generation, max_generation)
min_generation = (0 if old_root is None else old_root.generation) + 1
try:
is_valid: bool = await self._validate_batch(tree_id, to_check, min_generation)
except asyncio.CancelledError:
raise
except Exception as e:
self.log.error(f"Error in validate batch for {tree_id}: {e}")
is_valid = False
# If for some reason we have mismatched data using the light checks, recheck all history as a fallback.
if not is_valid:
self.log.warning(f"Light validation failed for {tree_id}. Validating all history.")
to_check = await self.wallet_rpc.dl_history(launcher_id=tree_id, min_generation=uint32(1))
# Already checked above.
self.log.info(
f"Validated chain hash {root.node_hash} in downloaded datastore. "
f"Wallet generation: {to_check[0].generation}"
)
to_check.pop(0)
is_valid = await self._validate_batch(tree_id, to_check, 0, max_generation)
try:
is_valid = await self._validate_batch(tree_id, to_check, 0)
except asyncio.CancelledError:
raise
except Exception as e:
self.log.error(f"Error in validate batch for {tree_id}: {e}")
is_valid = False
if not is_valid:
raise RuntimeError("Could not validate on-chain data.")
await self.data_store.set_validated_wallet_generation(tree_id, 0)
await self.data_store.rollback_to_generation(tree_id, 0)
raise RuntimeError("Could not validate on-chain data. Downloading from scratch as a fallback.")
self.log.info(
f"Finished downloading and validating {subscription.tree_id}. "
@@ -272,7 +323,9 @@ class DataLayer:
async def subscribe(self, store_id: bytes32, mode: DownloadMode, ip: str, port: uint16) -> None:
subscription = Subscription(store_id, mode, ip, port)
subscriptions = await self.get_subscriptions()
if subscription.tree_id in [subscription.tree_id for subscription in subscriptions]:
if subscription.tree_id in (subscription.tree_id for subscription in subscriptions):
await self.data_store.update_existing_subscription(subscription)
self.log.info(f"Successfully updated subscription {subscription.tree_id}")
return
await self.wallet_rpc.dl_track_new(subscription.tree_id)
async with self.subscription_lock:
@@ -281,8 +334,8 @@ class DataLayer:
async def unsubscribe(self, tree_id: bytes32) -> None:
subscriptions = await self.get_subscriptions()
if tree_id not in [subscription.tree_id for subscription in subscriptions]:
return
if tree_id not in (subscription.tree_id for subscription in subscriptions):
raise RuntimeError("No subscription found for the given tree_id.")
async with self.subscription_lock:
await self.data_store.unsubscribe(tree_id)
await self.wallet_rpc.dl_stop_tracking(tree_id)
@@ -292,6 +345,9 @@ class DataLayer:
async with self.subscription_lock:
return await self.data_store.get_subscriptions()
async def get_kv_diff(self, tree_id: bytes32, hash_1: bytes32, hash_2: bytes32) -> Set[DiffData]:
return await self.data_store.get_kv_diff(tree_id, hash_1, hash_2)
async def periodically_fetch_data(self) -> None:
fetch_data_interval = self.config.get("fetch_data_interval", 60)
while not self._shut_down:
chia/data_layer/data_layer_server.py
+22 -38
View File
@@ -1,8 +1,7 @@
import aiosqlite
import aiohttp
import json
# from random import Random
import logging
from typing import Any, Dict
from aiohttp import web # lgtm [py/import and import from]
from dataclasses import dataclass
@@ -11,15 +10,31 @@ from chia.data_layer.data_store import DataStore
from chia.util.db_wrapper import DBWrapper
from chia.types.blockchain_format.tree_hash import bytes32
from chia.data_layer.data_layer_types import TerminalNode, InsertionData
from chia.util.config import load_config
from chia.util.default_root import DEFAULT_ROOT_PATH
from chia.util.path import path_from_root, mkdir
# from tests.core.data_layer.util import generate_big_datastore
@dataclass
class DataLayerServer:
config: Dict[str, Any]
db_path: Path
log: logging.Logger
async def start(self) -> None:
self.log.info("Starting Data Layer Server.")
self.connection = await aiosqlite.connect(self.db_path)
self.db_wrapper = DBWrapper(self.connection)
self.data_store = await DataStore.create(db_wrapper=self.db_wrapper)
app = web.Application()
app.router.add_route("GET", "/ws", self.websocket_handler)
self.runner = web.AppRunner(app)
await self.runner.setup()
self.site = web.TCPSite(self.runner, self.config["host_ip"], port=self.config["host_port"])
await self.site.start()
self.log.info("Started Data Layer Server.")
async def stop(self) -> None:
self.log.info("Stopped Data Layer Server.")
await self.runner.cleanup()
async def handle_tree_root(self, request: Dict[str, str]) -> str:
tree_id = request["tree_id"]
requested_hash = request["node_hash"]
@@ -121,34 +136,3 @@ class DataLayerServer:
await ws.send_str(json_response)
return ws
async def start(self, config: Dict[Any, Any], db_path: Path) -> web.Application:
self.config = config
self.db_path = db_path
mkdir(self.db_path.parent)
self.connection = await aiosqlite.connect(self.db_path)
self.db_wrapper = DBWrapper(self.connection)
self.data_store = await DataStore.create(db_wrapper=self.db_wrapper)
"""
Uncomment if you need mock data, for testing purposes.
random = Random()
random.seed(100, version=2)
tree_id = bytes32(b"\0" * 32)
await self.data_store.create_tree(tree_id=tree_id)
await generate_big_datastore(data_store=self.data_store, tree_id=tree_id, random=random)
print("Generated datastore.")
"""
app = web.Application()
app.router.add_route("GET", "/ws", self.websocket_handler)
return app
if __name__ == "__main__":
config = load_config(DEFAULT_ROOT_PATH, "config.yaml", "data_layer")
db_path_replaced: str = config["database_path"].replace("CHALLENGE", config["selected_network"])
db_path = path_from_root(DEFAULT_ROOT_PATH, db_path_replaced)
data_layer_server = DataLayerServer()
web.run_app(data_layer_server.start(config, db_path), host=config["host_ip"], port=config["host_port"])
chia/data_layer/data_layer_wallet.py
+93 -5
View File
@@ -2,6 +2,7 @@ import logging
import json
import time
import dataclasses
from operator import attrgetter
from typing import Any, Optional, Tuple, Set, List, Dict, Type, TypeVar
from blspy import G2Element
@@ -48,6 +49,7 @@ class SingletonRecord(Streamable):
confirmed_at_height: uint32
lineage_proof: LineageProof
generation: uint32
timestamp: uint64
_T_DataLayerWallet = TypeVar("_T_DataLayerWallet", bound="DataLayerWallet")
@@ -164,6 +166,8 @@ class DataLayerWallet:
height: Optional[uint32] = None,
in_transaction: bool = False,
) -> None:
if await self.wallet_state_manager.dl_store.get_launcher(launcher_id) is not None:
return None
if spend is not None and spend.coin.name() == launcher_id: # spend.coin.name() == launcher_id is a sanity check
await self.new_launcher_spend(spend, height, in_transaction)
else:
@@ -207,6 +211,7 @@ class DataLayerWallet:
CoinSpend(launcher_coin, response.puzzle, response.solution),
height=response.height,
)
await self.wallet_state_manager.action_store.action_done(action_id)
async def new_launcher_spend(
self,
@@ -218,7 +223,6 @@ class DataLayerWallet:
if height is None:
height = (await self.get_launcher_coin_state(launcher_id)).spent_height
assert height is not None
full_puzhash, amount, root, inner_puzhash = launch_solution_to_singleton_info(
launcher_spend.solution.to_program()
)
@@ -231,10 +235,13 @@ class DataLayerWallet:
if ( # This is an unconfirmed singleton that we know about
singleton_record.coin_id == new_singleton.name() and not singleton_record.confirmed
):
await self.wallet_state_manager.dl_store.set_confirmed(singleton_record.coin_id, height)
timestamp = await self.wallet_state_manager.wallet_node.get_timestamp_for_height(height)
await self.wallet_state_manager.dl_store.set_confirmed(singleton_record.coin_id, height, timestamp)
else:
self.log.info(f"Spend of launcher {launcher_id} has already been processed")
return None
else:
timestamp = await self.wallet_state_manager.wallet_node.get_timestamp_for_height(height)
await self.wallet_state_manager.dl_store.add_singleton_record(
SingletonRecord(
coin_id=new_singleton.name(),
@@ -243,6 +250,7 @@ class DataLayerWallet:
inner_puzzle_hash=inner_puzhash,
confirmed=True,
confirmed_at_height=height,
timestamp=timestamp,
lineage_proof=LineageProof(
launcher_id,
create_host_layer_puzzle(inner_puzhash, root).get_tree_hash(),
@@ -388,6 +396,7 @@ class DataLayerWallet:
inner_puzzle_hash=inner_puzzle.get_tree_hash(),
confirmed=False,
confirmed_at_height=uint32(0),
timestamp=uint64(0),
lineage_proof=LineageProof(
launcher_coin.name(),
create_host_layer_puzzle(inner_puzzle.get_tree_hash(), initial_root).get_tree_hash(),
@@ -486,7 +495,7 @@ class DataLayerWallet:
created_at_time=uint64(int(time.time())),
to_puzzle_hash=next_inner_puzzle.get_tree_hash(),
amount=uint64(singleton_record.lineage_proof.amount),
fee_amount=uint64(0),
fee_amount=fee,
confirmed=False,
sent=uint32(10),
spend_bundle=spend_bundle,
@@ -518,6 +527,7 @@ class DataLayerWallet:
inner_puzzle_hash=next_inner_puzzle.get_tree_hash(),
confirmed=False,
confirmed_at_height=uint32(0),
timestamp=uint64(0),
lineage_proof=LineageProof(
singleton_record.coin_id,
next_inner_puzzle.get_tree_hash(),
@@ -603,6 +613,7 @@ class DataLayerWallet:
root=singleton_record.root,
confirmed=False,
confirmed_at_height=uint32(0),
timestamp=uint64(0),
inner_puzzle_hash=singleton_record.inner_puzzle_hash,
lineage_proof=LineageProof(
singleton_record.coin_id,
@@ -708,6 +719,7 @@ class DataLayerWallet:
return
new_singleton = Coin(parent_name, full_puzzle_hash, amount)
timestamp = await self.wallet_state_manager.wallet_node.get_timestamp_for_height(height)
await self.wallet_state_manager.dl_store.add_singleton_record(
SingletonRecord(
coin_id=new_singleton.name(),
@@ -716,6 +728,7 @@ class DataLayerWallet:
inner_puzzle_hash=inner_puzzle_hash,
confirmed=True,
confirmed_at_height=height,
timestamp=timestamp,
lineage_proof=LineageProof(
parent_name,
create_host_layer_puzzle(inner_puzzle_hash, root).get_tree_hash(),
@@ -736,6 +749,79 @@ class DataLayerWallet:
self.id(),
)
)
await self.potentially_handle_resubmit(singleton_record.launcher_id)
async def potentially_handle_resubmit(self, launcher_id: bytes32) -> None:
"""
This method is meant to detect a fork in our expected pending singletons and the singletons that have actually
been confirmed on chain. If there is a fork and the root on chain never changed, we will attempt to rebase our
singletons on to the new latest singleton. If there is a fork and the root changed, we assume that everything
has failed and delete any pending state.
"""
unconfirmed_singletons = await self.wallet_state_manager.dl_store.get_unconfirmed_singletons(launcher_id)
if len(unconfirmed_singletons) == 0:
return
unconfirmed_singletons = sorted(unconfirmed_singletons, key=attrgetter("generation"))
full_branch: List[SingletonRecord] = await self.wallet_state_manager.dl_store.get_all_singletons_for_launcher(
launcher_id,
min_generation=unconfirmed_singletons[0].generation,
)
if len(unconfirmed_singletons) == len(full_branch) and set(unconfirmed_singletons) == set(full_branch):
return
# Now we have detected a fork so we should check whether the root changed at all
parent_singleton = await self.wallet_state_manager.dl_store.get_singleton_record(
unconfirmed_singletons[0].lineage_proof.parent_name
)
if parent_singleton is None or any(parent_singleton.root != s.root for s in full_branch if s.confirmed):
root_changed: bool = True
else:
root_changed = False
# Regardless of whether the root changed or not, our old state is bad so let's eliminate it
# First let's find all of our txs matching our unconfirmed singletons
unconfirmed_ids: Set[bytes32] = {s.lineage_proof.parent_name for s in unconfirmed_singletons}
relevant_dl_txs: List[TransactionRecord] = []
for id in unconfirmed_ids:
tx = await self.wallet_state_manager.tx_store.get_transaction_record(id)
if tx is not None:
relevant_dl_txs.append(tx)
# Let's check our standard wallet for fee transactions related to these dl txs
all_spends: List[SpendBundle] = [tx.spend_bundle for tx in relevant_dl_txs if tx.spend_bundle is not None]
all_removal_ids: Set[bytes32] = {removal.name() for sb in all_spends for removal in sb.removals()}
unconfirmed_std_txs: List[
TransactionRecord
] = await self.wallet_state_manager.tx_store.get_unconfirmed_for_wallet(self.standard_wallet.id())
relevant_std_txs: List[TransactionRecord] = [
tx for tx in unconfirmed_std_txs if any(c.name() in all_removal_ids for c in tx.removals)
]
# Delete all of the relevant transactions
for tx in [*relevant_dl_txs, *relevant_std_txs]:
await self.wallet_state_manager.tx_store.delete_transaction_record(tx.name)
# Delete all of the unconfirmed singleton records
for singleton in unconfirmed_singletons:
await self.wallet_state_manager.dl_store.delete_singleton_record(singleton.coin_id)
if not root_changed:
# The root never changed so let's attempt a rebase
try:
all_txs: List[TransactionRecord] = []
for singleton in unconfirmed_singletons:
for tx in relevant_dl_txs:
if any(c.name() == singleton.coin_id for c in tx.additions):
if tx.spend_bundle is not None:
fee = uint64(tx.spend_bundle.fees())
else:
fee = uint64(0)
all_txs.extend(await self.create_update_state_spend(launcher_id, singleton.root, fee))
for tx in all_txs:
await self.wallet_state_manager.add_pending_transaction(tx)
except Exception as e:
self.log.warning(f"Something went wrong during attempted DL resubmit: {str(e)}")
# Something went wrong so let's delete anything pending that was created
for singleton in unconfirmed_singletons:
await self.wallet_state_manager.dl_store.delete_singleton_record(singleton.coin_id)
async def stop_tracking_singleton(self, launcher_id: bytes32) -> None:
await self.wallet_state_manager.dl_store.delete_singleton_records_by_launcher_id(launcher_id)
@@ -766,9 +852,11 @@ class DataLayerWallet:
# UTILITY #
###########
async def get_latest_singleton(self, launcher_id: bytes32) -> Optional[SingletonRecord]:
async def get_latest_singleton(
self, launcher_id: bytes32, only_confirmed: bool = False
) -> Optional[SingletonRecord]:
singleton: Optional[SingletonRecord] = await self.wallet_state_manager.dl_store.get_latest_singleton(
launcher_id
launcher_id, only_confirmed
)
return singleton
chia/data_layer/data_store.py
+33 -7
View File
@@ -444,16 +444,17 @@ class DataStore:
return roots
async def get_last_tree_root_by_hash(
self, tree_id: bytes32, hash: bytes32, max_generation: Optional[int] = None, *, lock: bool = True
self, tree_id: bytes32, hash: Optional[bytes32], max_generation: Optional[int] = None, *, lock: bool = True
) -> Optional[Root]:
async with self.db_wrapper.locked_transaction(lock=lock):
max_generation_str = f"AND generation < {max_generation} " if max_generation is not None else ""
node_hash_str = "AND node_hash == :node_hash " if hash is not None else "AND node_hash is NULL "
cursor = await self.db.execute(
"SELECT * FROM root WHERE tree_id == :tree_id "
f"{max_generation_str}"
"AND node_hash == :node_hash "
f"{node_hash_str}"
"ORDER BY generation DESC LIMIT 1",
{"tree_id": tree_id.hex(), "node_hash": hash.hex()},
{"tree_id": tree_id.hex(), "node_hash": None if hash is None else hash.hex()},
)
row = await cursor.fetchone()
@@ -1100,6 +1101,20 @@ class DataStore:
},
)
async def update_existing_subscription(self, subscription: Subscription, *, lock: bool = True) -> None:
async with self.db_wrapper.locked_transaction(lock=lock):
await self.db.execute(
"""
UPDATE subscriptions SET ip = :ip, port = :port, mode = :mode WHERE tree_id == :tree_id
""",
{
"tree_id": subscription.tree_id.hex(),
"mode": subscription.mode.value,
"ip": subscription.ip,
"port": subscription.port,
},
)
async def unsubscribe(self, tree_id: bytes32, *, lock: bool = True) -> None:
async with self.db_wrapper.locked_transaction(lock=lock):
await self.db.execute(
@@ -1111,6 +1126,13 @@ class DataStore:
{"tree_id": tree_id.hex()},
)
async def rollback_to_generation(self, tree_id: bytes32, target_generation: int, *, lock: bool = True) -> None:
async with self.db_wrapper.locked_transaction(lock=lock):
await self.db.execute(
"DELETE FROM root WHERE tree_id == :tree_id AND generation > :target_generation",
{"tree_id": tree_id.hex(), "target_generation": target_generation},
)
async def get_subscriptions(self, *, lock: bool = True) -> List[Subscription]:
subscriptions: List[Subscription] = []
@@ -1151,14 +1173,18 @@ class DataStore:
async def get_kv_diff(
self,
tree_id: bytes32,
hash_1: Optional[bytes32],
hash_2: Optional[bytes32],
hash_1: bytes32,
hash_2: bytes32,
*,
lock: bool = True,
) -> Set[DiffData]:
async with self.db_wrapper.locked_transaction(lock=lock):
old_pairs = set() if hash_1 is None else set(await self.get_keys_values(tree_id, hash_1, lock=False))
new_pairs = set() if hash_2 is None else set(await self.get_keys_values(tree_id, hash_2, lock=False))
old_pairs = set(await self.get_keys_values(tree_id, hash_1, lock=False))
new_pairs = set(await self.get_keys_values(tree_id, hash_2, lock=False))
if len(old_pairs) == 0 and hash_1 != bytes32([0] * 32):
return set()
if len(new_pairs) == 0 and hash_2 != bytes32([0] * 32):
return set()
insertions = set(
DiffData(type=OperationType.INSERT, key=node.key, value=node.value)
for node in new_pairs
chia/data_layer/dl_wallet_store.py
+53 -25
View File
@@ -14,6 +14,20 @@ from chia.wallet.lineage_proof import LineageProof
_T_DataLayerStore = TypeVar("_T_DataLayerStore", bound="DataLayerStore")
def _row_to_singleton_record(row: Row) -> SingletonRecord:
return SingletonRecord(
bytes32(row[0]),
bytes32(row[1]),
bytes32(row[2]),
bytes32(row[3]),
bool(row[4]),
uint32(row[5]),
LineageProof.from_bytes(row[6]),
uint32(row[7]),
uint64(row[8]),
)
class DataLayerStore:
"""
WalletUserStore keeps track of all user created wallets and necessary smart-contract data
@@ -38,7 +52,8 @@ class DataLayerStore:
" confirmed tinyint,"
" confirmed_at_height int,"
" proof blob,"
" generation int)" # This first singleton will be 0, then 1, and so on. This is handled by the DB.
" generation int," # This first singleton will be 0, then 1, and so on. This is handled by the DB.
" timestamp int)"
)
)
@@ -63,18 +78,6 @@ class DataLayerStore:
await cursor.close()
await self.db_connection.commit()
def _row_to_singleton_record(self, row: Row) -> SingletonRecord:
return SingletonRecord(
bytes32(row[0]),
bytes32(row[1]),
bytes32(row[2]),
bytes32(row[3]),
bool(row[4]),
uint32(row[5]),
LineageProof.from_bytes(row[6]),
uint32(row[7]),
)
async def add_singleton_record(self, record: SingletonRecord, in_transaction: bool) -> None:
"""
Store SingletonRecord in DB.
@@ -84,7 +87,7 @@ class DataLayerStore:
await self.db_wrapper.lock.acquire()
try:
cursor = await self.db_connection.execute(
"INSERT OR REPLACE INTO singleton_records VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
"INSERT OR REPLACE INTO singleton_records VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
record.coin_id,
record.launcher_id,
@@ -94,6 +97,7 @@ class DataLayerStore:
record.confirmed_at_height,
bytes(record.lineage_proof),
record.generation,
record.timestamp,
),
)
await cursor.close()
@@ -136,7 +140,7 @@ class DataLayerStore:
records = []
for row in rows:
records.append(self._row_to_singleton_record(row))
records.append(_row_to_singleton_record(row))
return records
@@ -151,25 +155,47 @@ class DataLayerStore:
row = await cursor.fetchone()
await cursor.close()
if row is not None:
return self._row_to_singleton_record(row)
return _row_to_singleton_record(row)
return None
async def get_latest_singleton(self, launcher_id: bytes32) -> Optional[SingletonRecord]:
async def get_latest_singleton(
self, launcher_id: bytes32, only_confirmed: bool = False
) -> Optional[SingletonRecord]:
"""
Checks DB for SingletonRecords with launcher_id: launcher_id and returns the most recent.
"""
# if tx_id in self.tx_record_cache:
# return self.tx_record_cache[tx_id]
cursor = await self.db_connection.execute(
"SELECT * from singleton_records WHERE launcher_id=? ORDER BY generation DESC LIMIT 1", (launcher_id,)
)
if only_confirmed:
# get latest confirmed root
cursor = await self.db_connection.execute(
"SELECT * from singleton_records WHERE launcher_id=? and confirmed = TRUE "
"ORDER BY generation DESC LIMIT 1",
(launcher_id,),
)
else:
cursor = await self.db_connection.execute(
"SELECT * from singleton_records WHERE launcher_id=? ORDER BY generation DESC LIMIT 1", (launcher_id,)
)
row = await cursor.fetchone()
await cursor.close()
if row is not None:
return self._row_to_singleton_record(row)
return _row_to_singleton_record(row)
return None
async def get_unconfirmed_singletons(self, launcher_id: bytes32) -> List[SingletonRecord]:
"""
Returns all singletons with a specific launcher id that have not yet been marked confirmed
"""
cursor = await self.db_connection.execute(
"SELECT * from singleton_records WHERE launcher_id=? AND confirmed=0", (launcher_id,)
)
rows = await cursor.fetchall()
await cursor.close()
records = [_row_to_singleton_record(row) for row in rows]
return records
async def get_singletons_by_root(self, launcher_id: bytes32, root: bytes32) -> List[SingletonRecord]:
cursor = await self.db_connection.execute(
"SELECT * from singleton_records WHERE launcher_id=? AND root=? ORDER BY generation DESC",
@@ -180,11 +206,11 @@ class DataLayerStore:
records = []
for row in rows:
records.append(self._row_to_singleton_record(row))
records.append(_row_to_singleton_record(row))
return records
async def set_confirmed(self, coin_id: bytes32, height: uint32) -> None:
async def set_confirmed(self, coin_id: bytes32, height: uint32, timestamp: uint64) -> None:
"""
Updates singleton record to be confirmed.
"""
@@ -192,7 +218,9 @@ class DataLayerStore:
if current is None or current.confirmed_at_height == height:
return
await self.add_singleton_record(dataclasses.replace(current, confirmed=True, confirmed_at_height=height), True)
await self.add_singleton_record(
dataclasses.replace(current, confirmed=True, confirmed_at_height=height, timestamp=timestamp), True
)
async def delete_singleton_record(self, coin_id: bytes32) -> None:
c = await self.db_connection.execute("DELETE FROM singleton_records WHERE coin_id=?", (coin_id,))
chia/data_layer/download_data.py
+4
View File
@@ -98,6 +98,10 @@ async def download_data_history(
root_json = json.loads(msg.data)
generation = root_json["generation"]
root = await data_store.get_tree_root(tree_id=tree_id, lock=lock)
# We've downloaded too much, rollback.
if root.generation > generation:
await data_store.rollback_to_generation(tree_id=tree_id, target_generation=generation, lock=lock)
return True
existing_generation = root.generation + 1
while existing_generation <= generation:
request = {
chia/pyinstaller.spec
+1
View File
@@ -54,6 +54,7 @@ version_data = copy_metadata(get_distribution("chia-blockchain"))[0]
block_cipher = None
SERVERS = [
"data_layer",
"wallet",
"full_node",
"harvester",
chia/rpc/data_layer_rpc_api.py
+59 -11
View File
@@ -1,6 +1,5 @@
import dataclasses
from typing import Any, Callable, Dict
from typing import Any, Callable, Dict, List
from chia.data_layer.data_layer import DataLayer
from chia.data_layer.data_layer_types import Side, DownloadMode
@@ -59,11 +58,14 @@ class DataLayerRpcApi:
"/get_keys_values": self.get_keys_values,
"/get_ancestors": self.get_ancestors,
"/get_root": self.get_root,
"/get_local_root": self.get_local_root,
"/get_roots": self.get_roots,
"/delete_key": self.delete_key,
"/insert": self.insert,
"/subscribe": self.subscribe,
"/unsubscribe": self.unsubscribe,
"/get_kv_diff": self.get_kv_diff,
"/get_root_history": self.get_root_history,
}
async def create_data_store(self, request: Dict[str, Any]) -> Dict[str, Any]:
@@ -159,7 +161,18 @@ class DataLayerRpcApi:
# todo input checks
if self.service is None:
raise Exception("Data layer not created")
res = await self.service.get_root(store_id)
rec = await self.service.get_root(store_id)
if rec is None:
raise Exception(f"Failed to get root for {store_id.hex()}")
return {"hash": rec.root, "confirmed": rec.confirmed, "timestamp": rec.timestamp}
async def get_local_root(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""get hash of latest tree root saved in our local datastore"""
store_id = bytes32(hexstr_to_bytes(request["id"]))
# todo input checks
if self.service is None:
raise Exception("Data layer not created")
res = await self.service.get_local_root(store_id)
return {"hash": res}
async def get_roots(self, request: Dict[str, Any]) -> Dict[str, Any]:
@@ -173,11 +186,12 @@ class DataLayerRpcApi:
roots = []
for id in store_ids:
id_bytes = bytes32.from_hexstr(id)
res = await self.service.get_root(store_id=id_bytes)
roots.append({"id": id_bytes, "hash": res})
rec = await self.service.get_root(id_bytes)
if rec is not None:
roots.append({"id": id_bytes, "hash": rec.root, "confirmed": rec.confirmed, "timestamp": rec.timestamp})
return {"root_hashes": roots}
async def subscribe(self, request: Dict[str, Any]) -> bool:
async def subscribe(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""
subscribe to singleton
"""
@@ -196,10 +210,11 @@ class DataLayerRpcApi:
mode = DownloadMode(req_mode)
if self.service is None:
raise Exception("Data layer not created")
await self.service.subscribe(store_id=store_id, mode=mode, ip=ip, port=port)
return True
store_id_bytes = bytes32.from_hexstr(store_id)
await self.service.subscribe(store_id=store_id_bytes, mode=mode, ip=ip, port=port)
return {}
async def unsubscribe(self, request: Dict[str, Any]) -> bool:
async def unsubscribe(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""
unsubscribe from singleton
"""
@@ -208,5 +223,38 @@ class DataLayerRpcApi:
raise Exception("missing store id in request")
if self.service is None:
raise Exception("Data layer not created")
await self.service.unsubscribe(store_id)
return True
store_id_bytes = bytes32.from_hexstr(store_id)
await self.service.unsubscribe(store_id_bytes)
return {}
async def get_root_history(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""
get history of state hashes for a store
"""
if self.service is None:
raise Exception("Data layer not created")
store_id = request["id"]
id_bytes = bytes32.from_hexstr(store_id)
records = await self.service.get_root_history(id_bytes)
res: List[Dict[str, Any]] = []
for rec in records:
res.insert(0, {"root_hash": rec.root, "confirmed": rec.confirmed, "timestamp": rec.timestamp})
return {"root_history": res}
async def get_kv_diff(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""
get kv diff between two root hashes
"""
if self.service is None:
raise Exception("Data layer not created")
store_id = request["id"]
id_bytes = bytes32.from_hexstr(store_id)
hash_1 = request["hash_1"]
hash_1_bytes = bytes32.from_hexstr(hash_1)
hash_2 = request["hash_2"]
hash_2_bytes = bytes32.from_hexstr(hash_2)
records = await self.service.get_kv_diff(id_bytes, hash_1_bytes, hash_2_bytes)
res: List[Dict[str, Any]] = []
for rec in records:
res.insert(0, {"type": rec.type.name, "key": rec.key.hex(), "value": rec.value.hex()})
return {"diff": res}
chia/rpc/data_layer_rpc_client.py
+16 -1
View File
@@ -38,6 +38,11 @@ class DataLayerRpcClient(RpcClient):
# TODO: better hinting for .fetch() (probably a TypedDict)
return response # type: ignore[no-any-return]
async def get_local_root(self, store_id: bytes32) -> Dict[str, Any]:
response = await self.fetch("get_local_root", {"id": store_id.hex()})
# TODO: better hinting for .fetch() (probably a TypedDict)
return response # type: ignore[no-any-return]
async def get_roots(self, store_ids: List[bytes32]) -> Dict[str, Any]:
response = await self.fetch("get_roots", {"ids": store_ids})
# TODO: better hinting for .fetch() (probably a TypedDict)
@@ -48,5 +53,15 @@ class DataLayerRpcClient(RpcClient):
return response # type: ignore[no-any-return]
async def unsubscribe(self, store_id: bytes32) -> Dict[str, Any]:
response = await self.fetch("unsubscribe", {"id": store_id})
response = await self.fetch("unsubscribe", {"id": store_id.hex()})
return response # type: ignore[no-any-return]
async def get_kv_diff(self, store_id: bytes32, hash_1: bytes32, hash_2: bytes32) -> Dict[str, Any]:
response = await self.fetch(
"get_kv_diff", {"id": store_id.hex(), "hash_1": hash_1.hex(), "hash_2": hash_2.hex()}
)
return response # type: ignore[no-any-return]
async def get_root_history(self, store_id: bytes32) -> Dict[str, Any]:
response = await self.fetch("get_root_history", {"id": store_id.hex()})
return response # type: ignore[no-any-return]
chia/rpc/wallet_rpc_api.py
+39 -33
View File
@@ -1344,14 +1344,15 @@ class WalletRpcApi:
)
try:
dl_tx, std_tx, launcher_id = await dl_wallet.generate_new_reporter(
bytes32.from_hexstr(request["root"]), fee=request.get("fee", uint64(0))
)
async with self.service.wallet_state_manager.lock:
dl_tx, std_tx, launcher_id = await dl_wallet.generate_new_reporter(
bytes32.from_hexstr(request["root"]), fee=request.get("fee", uint64(0))
)
await self.service.wallet_state_manager.add_pending_transaction(dl_tx)
await self.service.wallet_state_manager.add_pending_transaction(std_tx)
except ValueError as e:
log.error(f"Error while generating new reporter {e}")
return {"success": False, "error": str(e)}
await self.service.wallet_state_manager.add_pending_transaction(dl_tx)
await self.service.wallet_state_manager.add_pending_transaction(std_tx)
return {
"success": True,
@@ -1397,7 +1398,10 @@ class WalletRpcApi:
for _, wallet in self.service.wallet_state_manager.wallets.items():
if WalletType(wallet.type()) == WalletType.DATA_LAYER:
record = await wallet.get_latest_singleton(bytes32.from_hexstr(request["launcher_id"]))
only_confirmed = request.get("only_confirmed")
if only_confirmed is None:
only_confirmed = False
record = await wallet.get_latest_singleton(bytes32.from_hexstr(request["launcher_id"]), only_confirmed)
return {"singleton": None if record is None else record.to_json_dict()}
raise ValueError("No DataLayer wallet has been initialized")
@@ -1424,14 +1428,15 @@ class WalletRpcApi:
for _, wallet in self.service.wallet_state_manager.wallets.items():
if WalletType(wallet.type()) == WalletType.DATA_LAYER:
records = await wallet.create_update_state_spend(
bytes32.from_hexstr(request["launcher_id"]),
bytes32.from_hexstr(request["new_root"]),
fee=uint64(request.get("fee", 0)),
)
for record in records:
await self.service.wallet_state_manager.add_pending_transaction(record)
return {"tx_record": records[0].to_json_dict_convenience(self.service.config)}
async with self.service.wallet_state_manager.lock:
records = await wallet.create_update_state_spend(
bytes32.from_hexstr(request["launcher_id"]),
bytes32.from_hexstr(request["new_root"]),
fee=uint64(request.get("fee", 0)),
)
for record in records:
await self.service.wallet_state_manager.add_pending_transaction(record)
return {"tx_record": records[0].to_json_dict_convenience(self.service.config)}
raise ValueError("No DataLayer wallet has been initialized")
@@ -1442,25 +1447,26 @@ class WalletRpcApi:
for _, wallet in self.service.wallet_state_manager.wallets.items():
if WalletType(wallet.type()) == WalletType.DATA_LAYER:
# TODO: This method should optionally link the singletons with announcements.
# Otherwise spends are vulnerable to signature subtraction.
tx_records: List[TransactionRecord] = []
for launcher, root in request["updates"].items():
records = await wallet.create_update_state_spend(
bytes32.from_hexstr(launcher), bytes32.from_hexstr(root)
)
tx_records.extend(records)
# Now that we have all the txs, we need to aggregate them all into just one spend
modified_txs: List[TransactionRecord] = []
aggregate_spend = SpendBundle([], G2Element())
for tx in tx_records:
if tx.spend_bundle is not None:
aggregate_spend = SpendBundle.aggregate([aggregate_spend, tx.spend_bundle])
modified_txs.append(dataclasses.replace(tx, spend_bundle=None))
modified_txs[0] = dataclasses.replace(modified_txs[0], spend_bundle=aggregate_spend)
for tx in modified_txs:
await self.service.wallet_state_manager.add_pending_transaction(tx)
return {"tx_records": [rec.to_json_dict_convenience(self.service.config) for rec in modified_txs]}
async with self.service.wallet_state_manager.lock:
# TODO: This method should optionally link the singletons with announcements.
# Otherwise spends are vulnerable to signature subtraction.
tx_records: List[TransactionRecord] = []
for launcher, root in request["updates"].items():
records = await wallet.create_update_state_spend(
bytes32.from_hexstr(launcher), bytes32.from_hexstr(root)
)
tx_records.extend(records)
# Now that we have all the txs, we need to aggregate them all into just one spend
modified_txs: List[TransactionRecord] = []
aggregate_spend = SpendBundle([], G2Element())
for tx in tx_records:
if tx.spend_bundle is not None:
aggregate_spend = SpendBundle.aggregate([aggregate_spend, tx.spend_bundle])
modified_txs.append(dataclasses.replace(tx, spend_bundle=None))
modified_txs[0] = dataclasses.replace(modified_txs[0], spend_bundle=aggregate_spend)
for tx in modified_txs:
await self.service.wallet_state_manager.add_pending_transaction(tx)
return {"tx_records": [rec.to_json_dict_convenience(self.service.config) for rec in modified_txs]}
raise ValueError("No DataLayer wallet has been initialized")
chia/rpc/wallet_rpc_client.py
+4 -2
View File
@@ -489,8 +489,10 @@ class WalletRpcClient(RpcClient):
await self.fetch("dl_stop_tracking", request)
return None
async def dl_latest_singleton(self, launcher_id: bytes32) -> Optional[SingletonRecord]:
request = {"launcher_id": launcher_id.hex()}
async def dl_latest_singleton(
self, launcher_id: bytes32, only_confirmed: bool = False
) -> Optional[SingletonRecord]:
request = {"launcher_id": launcher_id.hex(), "only_confirmed": only_confirmed}
response = await self.fetch("dl_latest_singleton", request)
return None if response["singleton"] is None else SingletonRecord.from_json_dict(response["singleton"])
chia/util/initial-config.yaml
+3 -1
View File
@@ -516,8 +516,10 @@ data_layer:
database_path: "data_layer/db/data_layer_CHALLENGE.sqlite"
# Data for running a data layer server.
host_ip: 127.0.0.1
host_ip: 0.0.0.0
host_port: 8000
# Switch this to True if we want to run the server.
run_server: False
# Data for running a data layer client.
fetch_data_interval: 60
selected_network: *selected_network
chia/wallet/puzzles/singleton_top_layer_v1_1.clvm
+9 -13
View File
@@ -11,14 +11,14 @@
; It takes its singleton identity from a coin with a launcher puzzle which guarantees that it is unique.
(include condition_codes.clvm)
(include curry-and-treehash.clinc)
(include curry-and-treehash.clinc) ; also imports the constant ONE == 1
(include singleton_truths.clib)
; takes a lisp tree and returns the hash of it
(defun sha256tree1 (TREE)
(if (l TREE)
(sha256 2 (sha256tree1 (f TREE)) (sha256tree1 (r TREE)))
(sha256 1 TREE)
(sha256 ONE TREE)
)
)
@@ -48,7 +48,7 @@
)
; assembles information from the solution to create our own full ID including asserting our parent is a singleton
(defun create_my_ID (SINGLETON_STRUCT full_puzzle_hash parent_parent parent_inner_puzzle_hash parent_amount my_amount)
(defun-inline create_my_ID (SINGLETON_STRUCT full_puzzle_hash parent_parent parent_inner_puzzle_hash parent_amount my_amount)
(sha256 (sha256 parent_parent (calculate_full_puzzle_hash SINGLETON_STRUCT parent_inner_puzzle_hash) parent_amount)
full_puzzle_hash
my_amount)
@@ -80,7 +80,7 @@
;; Returns a (bool . bool)
(defun odd_cons_m113 (output_amount)
(c
(= (logand output_amount 1) 1) ;; is it odd?
(= (logand output_amount ONE) ONE) ;; is it odd?
(= output_amount -113) ;; is it the escape value?
)
)
@@ -124,17 +124,16 @@
; this final stager asserts our ID
; it also runs the innerpuz with the innersolution with the "truths" added
; it then passes that output conditions from the innerpuz to the morph conditions function
(defun stager_three (SINGLETON_STRUCT lineage_proof my_id full_puzhash innerpuzhash my_amount INNER_PUZZLE inner_solution)
(defun-inline stager_three (SINGLETON_STRUCT my_id INNER_PUZZLE inner_solution)
(c (list ASSERT_MY_COIN_ID my_id) (check_and_morph_conditions_for_singleton SINGLETON_STRUCT (a INNER_PUZZLE inner_solution) 0))
)
; this checks whether we are an eve spend or not and calculates our full coin ID appropriately and passes it on to the final stager
; if we are the eve spend it also adds the additional checks that our parent's puzzle is the standard launcher format and that out parent ID is the same as our singleton ID
(defun stager_two (SINGLETON_STRUCT lineage_proof full_puzhash innerpuzhash my_amount INNER_PUZZLE inner_solution)
(defun-inline stager_two (SINGLETON_STRUCT lineage_proof full_puzhash my_amount INNER_PUZZLE inner_solution)
(stager_three
SINGLETON_STRUCT
lineage_proof
(if (is_not_eve_proof lineage_proof)
(create_my_ID
SINGLETON_STRUCT
@@ -152,17 +151,14 @@
(x)
)
)
full_puzhash
innerpuzhash
my_amount
INNER_PUZZLE
inner_solution
)
)
; this calculates our current full puzzle hash and passes it to stager two
(defun stager_one (SINGLETON_STRUCT lineage_proof my_innerpuzhash my_amount INNER_PUZZLE inner_solution)
(stager_two SINGLETON_STRUCT lineage_proof (calculate_full_puzzle_hash SINGLETON_STRUCT my_innerpuzhash) my_innerpuzhash my_amount INNER_PUZZLE inner_solution)
(defun-inline stager_one (SINGLETON_STRUCT lineage_proof my_innerpuzhash my_amount INNER_PUZZLE inner_solution)
(stager_two SINGLETON_STRUCT lineage_proof (calculate_full_puzzle_hash SINGLETON_STRUCT my_innerpuzhash) my_amount INNER_PUZZLE inner_solution)
)
@@ -170,7 +166,7 @@
; if our value is not an odd amount then we are invalid
; this calculates my_innerpuzhash and passes all values to stager_one
(if (logand my_amount 1)
(if (logand my_amount ONE)
(stager_one SINGLETON_STRUCT lineage_proof (sha256tree1 INNER_PUZZLE) my_amount INNER_PUZZLE inner_solution)
(x)
)
chia/wallet/puzzles/singleton_top_layer_v1_1.clvm.hex
+1 -1
View File
@@ -1 +1 @@
ff02ffff01ff02ffff03ffff18ff2fffff010180ffff01ff02ff36ffff04ff02ffff04ff05ffff04ff17ffff04ffff02ff26ffff04ff02ffff04ff0bff80808080ffff04ff2fffff04ff0bffff04ff5fff808080808080808080ffff01ff088080ff0180ffff04ffff01ffffffff4602ff3304ffff0101ff02ffff02ffff03ff05ffff01ff02ff5cffff04ff02ffff04ff0dffff04ffff0bff2cffff0bff24ff3880ffff0bff2cffff0bff2cffff0bff24ff3480ff0980ffff0bff2cff0bffff0bff24ff8080808080ff8080808080ffff010b80ff0180ff02ffff03ff0bffff01ff02ff32ffff04ff02ffff04ff05ffff04ff0bffff04ff17ffff04ffff02ff2affff04ff02ffff04ffff02ffff03ffff09ff23ff2880ffff0181b3ff8080ff0180ff80808080ff80808080808080ffff01ff02ffff03ff17ff80ffff01ff088080ff018080ff0180ffffffff0bffff0bff17ffff02ff3affff04ff02ffff04ff09ffff04ff2fffff04ffff02ff26ffff04ff02ffff04ff05ff80808080ff808080808080ff5f80ff0bff81bf80ff02ffff03ffff20ffff22ff4fff178080ffff01ff02ff7effff04ff02ffff04ff6fffff04ffff04ffff02ffff03ff4fffff01ff04ff23ffff04ffff02ff3affff04ff02ffff04ff09ffff04ff53ffff04ffff02ff26ffff04ff02ffff04ff05ff80808080ff808080808080ff738080ffff011380ff0180ffff02ff7cffff04ff02ffff04ff05ffff04ff1bffff04ffff21ff4fff1780ff80808080808080ff8080808080ffff01ff088080ff0180ffff04ffff09ffff18ff05ffff010180ffff010180ffff09ff05ffff01818f8080ff0bff2cffff0bff24ff3080ffff0bff2cffff0bff2cffff0bff24ff3480ff0580ffff0bff2cffff02ff5cffff04ff02ffff04ff07ffff04ffff0bff24ff2480ff8080808080ffff0bff24ff8080808080ffffff02ffff03ffff07ff0580ffff01ff0bffff0102ffff02ff26ffff04ff02ffff04ff09ff80808080ffff02ff26ffff04ff02ffff04ff0dff8080808080ffff01ff0bffff0101ff058080ff0180ff02ff5effff04ff02ffff04ff05ffff04ff0bffff04ffff02ff3affff04ff02ffff04ff09ffff04ff17ffff04ffff02ff26ffff04ff02ffff04ff05ff80808080ff808080808080ffff04ff17ffff04ff2fffff04ff5fffff04ff81bfff80808080808080808080ffff04ffff04ff20ffff04ff17ff808080ffff02ff7cffff04ff02ffff04ff05ffff04ffff02ff82017fff8202ff80ffff01ff80808080808080ffff02ff2effff04ff02ffff04ff05ffff04ff0bffff04ffff02ffff03ff3bffff01ff02ff22ffff04ff02ffff04ff05ffff04ff17ffff04ff13ffff04ff2bffff04ff5bffff04ff5fff808080808080808080ffff01ff02ffff03ffff09ff15ffff0bff13ff1dff2b8080ffff01ff0bff15ff17ff5f80ffff01ff088080ff018080ff0180ffff04ff17ffff04ff2fffff04ff5fffff04ff81bfffff04ff82017fff8080808080808080808080ff02ffff03ff05ffff011bffff010b80ff0180ff018080
ff02ffff01ff02ffff03ffff18ff2fff3480ffff01ff04ffff04ff10ffff04ffff02ffff03ff77ffff01ff0bffff0bff27ffff02ff36ffff04ff02ffff04ff09ffff04ff57ffff04ffff02ff2effff04ff02ffff04ff05ff80808080ff808080808080ff81b780ffff02ff36ffff04ff02ffff04ff09ffff04ffff02ff2effff04ff02ffff04ff0bff80808080ffff04ffff02ff2effff04ff02ffff04ff05ff80808080ff808080808080ff2f80ffff01ff02ffff03ffff09ff15ffff0bff27ff1dff578080ffff01ff0bff15ffff02ff36ffff04ff02ffff04ff09ffff04ffff02ff2effff04ff02ffff04ff0bff80808080ffff04ffff02ff2effff04ff02ffff04ff05ff80808080ff808080808080ff2f80ffff01ff088080ff018080ff0180ff808080ffff02ff2affff04ff02ffff04ff05ffff04ffff02ff0bff5f80ffff01ff80808080808080ffff01ff088080ff0180ffff04ffff01ffffff46ff0233ffff0401ff0102ffffff02ffff03ff05ffff01ff02ff12ffff04ff02ffff04ff0dffff04ffff0bff3cffff0bff34ff2480ffff0bff3cffff0bff3cffff0bff34ff2c80ff0980ffff0bff3cff0bffff0bff34ff8080808080ff8080808080ffff010b80ff0180ffff02ffff03ff0bffff01ff02ff3affff04ff02ffff04ff05ffff04ff0bffff04ff17ffff04ffff02ff26ffff04ff02ffff04ffff02ffff03ffff09ff23ff3880ffff0181b3ff8080ff0180ff80808080ff80808080808080ffff01ff02ffff03ff17ff80ffff01ff088080ff018080ff0180ff02ffff03ffff20ffff22ff4fff178080ffff01ff02ff3effff04ff02ffff04ff6fffff04ffff04ffff02ffff03ff4fffff01ff04ff23ffff04ffff02ff36ffff04ff02ffff04ff09ffff04ff53ffff04ffff02ff2effff04ff02ffff04ff05ff80808080ff808080808080ff738080ffff011380ff0180ffff02ff2affff04ff02ffff04ff05ffff04ff1bffff04ffff21ff4fff1780ff80808080808080ff8080808080ffff01ff088080ff0180ffffff04ffff09ffff18ff05ff3480ff3480ffff09ff05ffff01818f8080ff0bff3cffff0bff34ff2880ffff0bff3cffff0bff3cffff0bff34ff2c80ff0580ffff0bff3cffff02ff12ffff04ff02ffff04ff07ffff04ffff0bff34ff3480ff8080808080ffff0bff34ff8080808080ffff02ffff03ffff07ff0580ffff01ff0bffff0102ffff02ff2effff04ff02ffff04ff09ff80808080ffff02ff2effff04ff02ffff04ff0dff8080808080ffff01ff0bff34ff058080ff0180ff02ffff03ff05ffff011bffff010b80ff0180ff018080
chia/wallet/puzzles/singleton_top_layer_v1_1.clvm.hex.sha256tree
+1 -1
View File
@@ -1 +1 @@
ac512e3c101d139a75447b05abde22dff3ce5ebc28b49157ca09b51d60040c4f
ffc0846eea5898e9aefea956b7d0e08df97a1c080910648270d979a4900d88aa
pyproject.toml
+1
View File
@@ -3,6 +3,7 @@ requires = ["setuptools>=42", "wheel", "setuptools_scm[toml]>=4.1.2"]
build-backend = "setuptools.build_meta"
[tool.setuptools_scm]
fallback_version = "unknown-no-.git-directory"
local_scheme = "no-local-version"
[tool.black]
setup.py
+2 -2
View File
@@ -70,7 +70,6 @@ kwargs = dict(
python_requires=">=3.7, <4",
keywords="chia blockchain node",
install_requires=dependencies,
setup_requires=["setuptools_scm"],
extras_require=dict(
uvloop=["uvloop"],
dev=dev_dependencies,
@@ -102,6 +101,8 @@ kwargs = dict(
"chia.types",
"chia.util",
"chia.wallet",
"chia.wallet.db_wallet",
"chia.wallet.dlo_wallet",
"chia.wallet.puzzles",
"chia.wallet.rl_wallet",
"chia.wallet.cat_wallet",
@@ -136,7 +137,6 @@ kwargs = dict(
"chia.ssl": ["chia_ca.crt", "chia_ca.key", "dst_root_ca.pem"],
"mozilla-ca": ["cacert.pem"],
},
use_scm_version={"fallback_version": "unknown-no-.git-directory"},
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
zip_safe=False,
tests/core/data_layer/test_data_rpc.py
+195 -35
View File
@@ -1,17 +1,15 @@
import asyncio
from pathlib import Path
from shutil import rmtree
from typing import AsyncIterator, Dict, List, Tuple, Any
from typing import AsyncIterator, Dict, List, Tuple, Set
import pytest
# flake8: noqa: F401
from chia.consensus.block_rewards import calculate_base_farmer_reward, calculate_pool_reward
from chia.data_layer.data_layer import DataLayer
from chia.data_layer.data_layer_types import DiffData, OperationType
from chia.rpc.data_layer_rpc_api import DataLayerRpcApi
from chia.rpc.rpc_server import start_rpc_server
from chia.rpc.wallet_rpc_api import WalletRpcApi
from chia.rpc.wallet_rpc_client import WalletRpcClient
from chia.server.server import ChiaServer
from chia.server.start_data_layer import service_kwargs_for_data_layer
from chia.server.start_service import Service
from chia.simulator.full_node_simulator import FullNodeSimulator
@@ -19,12 +17,9 @@ from chia.simulator.simulator_protocol import FarmNewBlockProtocol
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.types.peer_info import PeerInfo
from chia.util.byte_types import hexstr_to_bytes
from chia.util.config import load_config
from chia.util.ints import uint16, uint32
from chia.wallet.transaction_record import TransactionRecord
from chia.wallet.wallet_node import WalletNode
from tests.core.data_layer.util import ChiaRoot
from tests.setup_nodes import setup_simulators_and_wallets, self_hostname, bt, _teardown_nodes
from tests.setup_nodes import setup_simulators_and_wallets, bt
from tests.time_out_assert import time_out_assert
from tests.wallet.rl_wallet.test_rl_rpc import is_transaction_confirmed
@@ -54,16 +49,13 @@ async def one_wallet_node() -> AsyncIterator[nodes]:
async def one_wallet_node_and_rpc() -> AsyncIterator[nodes]:
async for nodes in setup_simulators_and_wallets(1, 1, {}):
full_nodes, wallets = nodes
full_node_api = full_nodes[0]
wallet_node_0, wallet_server_0 = wallets[0]
api_user = WalletRpcApi(wallet_node_0)
config = bt.config
hostname = config["self_hostname"]
daemon_port = config["daemon_port"]
test_rpc_port = uint16(21529)
rpc_cleanup = await start_rpc_server(
api_user,
WalletRpcApi(wallet_node_0),
hostname,
daemon_port,
test_rpc_port,
@@ -72,12 +64,7 @@ async def one_wallet_node_and_rpc() -> AsyncIterator[nodes]:
config,
connect_to_daemon=False,
)
client = await WalletRpcClient.create(self_hostname, test_rpc_port, bt.root_path, config)
yield wallet_node_0, full_node_api
client.close()
await client.await_closed()
yield wallet_node_0, full_nodes[0]
await rpc_cleanup()
@@ -85,8 +72,6 @@ async def one_wallet_node_and_rpc() -> AsyncIterator[nodes]:
async def test_create_insert_get(one_wallet_node_and_rpc: nodes) -> None:
root_path = bt.root_path
wallet_node, full_node_api = one_wallet_node_and_rpc
config = load_config(root_path, "config.yaml")
config["data_layer"]["database_path"] = "data_layer_test.sqlite"
num_blocks = 15
assert wallet_node.server
await wallet_node.server.start_client(PeerInfo("localhost", uint16(full_node_api.server._port)), None)
@@ -119,6 +104,9 @@ async def test_create_insert_get(one_wallet_node_and_rpc: nodes) -> None:
await asyncio.sleep(0.2)
await time_out_assert(15, is_transaction_confirmed, True, "this is unused", wallet_rpc_api, update_tx_rec0)
res = await data_rpc_api.get_value({"id": store_id.hex(), "key": key.hex()})
wallet_root = await data_rpc_api.get_root({"id": store_id.hex()})
local_root = await data_rpc_api.get_local_root({"id": store_id.hex()})
assert wallet_root["hash"] == local_root["hash"]
assert hexstr_to_bytes(res["value"]) == value
changelist = [{"action": "delete", "key": key.hex()}]
res = await data_rpc_api.batch_update({"id": store_id.hex(), "changelist": changelist})
@@ -130,14 +118,16 @@ async def test_create_insert_get(one_wallet_node_and_rpc: nodes) -> None:
await time_out_assert(15, is_transaction_confirmed, True, "this is unused", wallet_rpc_api, update_tx_rec1)
with pytest.raises(Exception):
val = await data_rpc_api.get_value({"id": store_id.hex(), "key": key.hex()})
wallet_root = await data_rpc_api.get_root({"id": store_id.hex()})
local_root = await data_rpc_api.get_local_root({"id": store_id.hex()})
assert wallet_root["hash"] == bytes32([0] * 32)
assert local_root["hash"] == None
@pytest.mark.asyncio
async def test_create_double_insert(one_wallet_node_and_rpc: nodes) -> None:
root_path = bt.root_path
wallet_node, full_node_api = one_wallet_node_and_rpc
config = load_config(root_path, "config.yaml")
config["data_layer"]["database_path"] = "data_layer_test.sqlite"
num_blocks = 15
assert wallet_node.server
await wallet_node.server.start_client(PeerInfo("localhost", uint16(full_node_api.server._port)), None)
@@ -200,8 +190,6 @@ async def test_create_double_insert(one_wallet_node_and_rpc: nodes) -> None:
async def test_keys_values_ancestors(one_wallet_node_and_rpc: nodes) -> None:
root_path = bt.root_path
wallet_node, full_node_api = one_wallet_node_and_rpc
config = load_config(root_path, "config.yaml")
config["data_layer"]["database_path"] = "data_layer_test.sqlite"
num_blocks = 15
assert wallet_node.server
await wallet_node.server.start_client(PeerInfo("localhost", uint16(full_node_api.server._port)), None)
@@ -258,7 +246,9 @@ async def test_keys_values_ancestors(one_wallet_node_and_rpc: nodes) -> None:
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
res_before = await data_rpc_api.get_roots({"ids": [store_id.hex()]})
res_before = await data_rpc_api.get_root({"id": store_id.hex()})
assert res_before["confirmed"] is True
assert res_before["timestamp"] > 0
key6 = b"tasdfsd"
value6 = b"\x08\x02"
changelist = [{"action": "insert", "key": key6.hex(), "value": value6.hex()}]
@@ -272,13 +262,11 @@ async def test_keys_values_ancestors(one_wallet_node_and_rpc: nodes) -> None:
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.2)
await time_out_assert(15, is_transaction_confirmed, True, "this is unused", wallet_rpc_api, update_tx_rec0)
res_after = await data_rpc_api.get_roots({"ids": [store_id.hex()]})
pairs_before = await data_rpc_api.get_keys_values(
{"id": store_id.hex(), "root_hash": res_before["root_hashes"][0]["hash"].hex()}
)
pairs_after = await data_rpc_api.get_keys_values(
{"id": store_id.hex(), "root_hash": res_after["root_hashes"][0]["hash"].hex()}
)
res_after = await data_rpc_api.get_root({"id": store_id.hex()})
assert res_after["confirmed"] is True
assert res_after["timestamp"] > res_before["timestamp"]
pairs_before = await data_rpc_api.get_keys_values({"id": store_id.hex(), "root_hash": res_before["hash"].hex()})
pairs_after = await data_rpc_api.get_keys_values({"id": store_id.hex(), "root_hash": res_after["hash"].hex()})
assert len(pairs_before["keys_values"]) == 5
assert len(pairs_after["keys_values"]) == 7
@@ -287,8 +275,6 @@ async def test_keys_values_ancestors(one_wallet_node_and_rpc: nodes) -> None:
async def test_get_roots(one_wallet_node_and_rpc: nodes) -> None:
root_path = bt.root_path
wallet_node, full_node_api = one_wallet_node_and_rpc
config = load_config(root_path, "config.yaml")
config["data_layer"]["database_path"] = "data_layer_test.sqlite"
num_blocks = 15
assert wallet_node.server
await wallet_node.server.start_client(PeerInfo("localhost", uint16(full_node_api.server._port)), None)
@@ -336,7 +322,9 @@ async def test_get_roots(one_wallet_node_and_rpc: nodes) -> None:
await time_out_assert(15, is_transaction_confirmed, True, "this is unused", wallet_rpc_api, update_tx_rec0)
roots = await data_rpc_api.get_roots({"ids": [store_id1.hex(), store_id2.hex()]})
assert roots["root_hashes"][1]["id"] == store_id2
assert roots["root_hashes"][1]["hash"] == None
assert roots["root_hashes"][1]["hash"] == bytes32([0] * 32)
assert roots["root_hashes"][1]["confirmed"] is True
assert roots["root_hashes"][1]["timestamp"] > 0
key4 = b"d"
value4 = b"\x06\x03"
changelist = [{"action": "insert", "key": key4.hex(), "value": value4.hex()}]
@@ -353,6 +341,178 @@ async def test_get_roots(one_wallet_node_and_rpc: nodes) -> None:
roots = await data_rpc_api.get_roots({"ids": [store_id1.hex(), store_id2.hex()]})
assert roots["root_hashes"][1]["id"] == store_id2
assert roots["root_hashes"][1]["hash"] is not None
assert roots["root_hashes"][1]["hash"] != bytes32([0] * 32)
assert roots["root_hashes"][1]["confirmed"] is True
assert roots["root_hashes"][1]["timestamp"] > 0
@pytest.mark.asyncio
async def test_get_root_history(one_wallet_node_and_rpc: nodes) -> None:
root_path = bt.root_path
wallet_node, full_node_api = one_wallet_node_and_rpc
num_blocks = 15
assert wallet_node.server
await wallet_node.server.start_client(PeerInfo("localhost", uint16(full_node_api.server._port)), None)
assert wallet_node.wallet_state_manager is not None
ph = await wallet_node.wallet_state_manager.main_wallet.get_new_puzzlehash()
for i in range(0, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.5)
funds = sum(
[calculate_pool_reward(uint32(i)) + calculate_base_farmer_reward(uint32(i)) for i in range(1, num_blocks)]
)
await time_out_assert(15, wallet_node.wallet_state_manager.main_wallet.get_confirmed_balance, funds)
wallet_rpc_api = WalletRpcApi(wallet_node)
async for data_layer in init_data_layer(root_path):
data_rpc_api = DataLayerRpcApi(data_layer)
res = await data_rpc_api.create_data_store({})
assert res is not None
store_id1 = bytes32(hexstr_to_bytes(res["id"]))
for i in range(0, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.2)
res = await data_rpc_api.create_data_store({})
assert res is not None
key1 = b"a"
value1 = b"\x01\x02"
changelist: List[Dict[str, str]] = [{"action": "insert", "key": key1.hex(), "value": value1.hex()}]
key2 = b"b"
value2 = b"\x03\x02"
changelist.append({"action": "insert", "key": key2.hex(), "value": value2.hex()})
key3 = b"c"
value3 = b"\x04\x05"
changelist.append({"action": "insert", "key": key3.hex(), "value": value3.hex()})
res = await data_rpc_api.batch_update({"id": store_id1.hex(), "changelist": changelist})
update_tx_rec0 = res["tx_id"]
await asyncio.sleep(1)
for i in range(0, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.2)
await time_out_assert(15, is_transaction_confirmed, True, "this is unused", wallet_rpc_api, update_tx_rec0)
history1 = await data_rpc_api.get_root_history({"id": store_id1.hex()})
assert len(history1["root_history"]) == 2
assert history1["root_history"][0]["root_hash"] == bytes32([0] * 32)
assert history1["root_history"][0]["confirmed"] is True
assert history1["root_history"][0]["timestamp"] > 0
assert history1["root_history"][1]["root_hash"] != bytes32([0] * 32)
assert history1["root_history"][1]["confirmed"] is True
assert history1["root_history"][1]["timestamp"] > 0
key4 = b"d"
value4 = b"\x06\x03"
changelist = [{"action": "insert", "key": key4.hex(), "value": value4.hex()}]
key5 = b"e"
value5 = b"\x07\x01"
changelist.append({"action": "insert", "key": key5.hex(), "value": value5.hex()})
res = await data_rpc_api.batch_update({"id": store_id1.hex(), "changelist": changelist})
update_tx_rec1 = res["tx_id"]
await asyncio.sleep(1)
for i in range(0, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.2)
await time_out_assert(15, is_transaction_confirmed, True, "this is unused", wallet_rpc_api, update_tx_rec1)
history2 = await data_rpc_api.get_root_history({"id": store_id1.hex()})
assert len(history2["root_history"]) == 3
assert history2["root_history"][0]["root_hash"] == bytes32([0] * 32)
assert history2["root_history"][0]["confirmed"] is True
assert history2["root_history"][0]["timestamp"] > 0
assert history2["root_history"][1]["root_hash"] == history1["root_history"][1]["root_hash"]
assert history2["root_history"][1]["confirmed"] is True
assert history2["root_history"][1]["timestamp"] > history2["root_history"][0]["timestamp"]
assert history2["root_history"][2]["confirmed"] is True
assert history2["root_history"][2]["timestamp"] > history2["root_history"][1]["timestamp"]
@pytest.mark.asyncio
async def test_get_kv_diff(one_wallet_node_and_rpc: nodes) -> None:
root_path = bt.root_path
wallet_node, full_node_api = one_wallet_node_and_rpc
num_blocks = 15
assert wallet_node.server
await wallet_node.server.start_client(PeerInfo("localhost", uint16(full_node_api.server._port)), None)
assert wallet_node.wallet_state_manager is not None
ph = await wallet_node.wallet_state_manager.main_wallet.get_new_puzzlehash()
for i in range(0, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.5)
funds = sum(
[calculate_pool_reward(uint32(i)) + calculate_base_farmer_reward(uint32(i)) for i in range(1, num_blocks)]
)
await time_out_assert(15, wallet_node.wallet_state_manager.main_wallet.get_confirmed_balance, funds)
wallet_rpc_api = WalletRpcApi(wallet_node)
async for data_layer in init_data_layer(root_path):
data_rpc_api = DataLayerRpcApi(data_layer)
res = await data_rpc_api.create_data_store({})
assert res is not None
store_id1 = bytes32(hexstr_to_bytes(res["id"]))
for i in range(0, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.2)
res = await data_rpc_api.create_data_store({})
assert res is not None
key1 = b"a"
value1 = b"\x01\x02"
changelist: List[Dict[str, str]] = [{"action": "insert", "key": key1.hex(), "value": value1.hex()}]
key2 = b"b"
value2 = b"\x03\x02"
changelist.append({"action": "insert", "key": key2.hex(), "value": value2.hex()})
key3 = b"c"
value3 = b"\x04\x05"
changelist.append({"action": "insert", "key": key3.hex(), "value": value3.hex()})
res = await data_rpc_api.batch_update({"id": store_id1.hex(), "changelist": changelist})
update_tx_rec0 = res["tx_id"]
await asyncio.sleep(1)
for i in range(0, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.2)
await time_out_assert(15, is_transaction_confirmed, True, "this is unused", wallet_rpc_api, update_tx_rec0)
history = await data_rpc_api.get_root_history({"id": store_id1.hex()})
diff_res = await data_rpc_api.get_kv_diff(
{
"id": store_id1.hex(),
"hash_1": bytes32([0] * 32).hex(),
"hash_2": history["root_history"][1]["root_hash"].hex(),
}
)
assert len(diff_res["diff"]) == 3
diff1 = {"type": "INSERT", "key": key1.hex(), "value": value1.hex()}
diff2 = {"type": "INSERT", "key": key2.hex(), "value": value2.hex()}
diff3 = {"type": "INSERT", "key": key3.hex(), "value": value3.hex()}
assert diff1 in diff_res["diff"]
assert diff2 in diff_res["diff"]
assert diff3 in diff_res["diff"]
key4 = b"d"
value4 = b"\x06\x03"
changelist = [{"action": "insert", "key": key4.hex(), "value": value4.hex()}]
key5 = b"e"
value5 = b"\x07\x01"
changelist.append({"action": "insert", "key": key5.hex(), "value": value5.hex()})
changelist.append({"action": "delete", "key": key1.hex()})
res = await data_rpc_api.batch_update({"id": store_id1.hex(), "changelist": changelist})
update_tx_rec1 = res["tx_id"]
await asyncio.sleep(1)
for i in range(0, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await asyncio.sleep(0.2)
await time_out_assert(15, is_transaction_confirmed, True, "this is unused", wallet_rpc_api, update_tx_rec1)
history = await data_rpc_api.get_root_history({"id": store_id1.hex()})
diff_res = await data_rpc_api.get_kv_diff(
{
"id": store_id1.hex(),
"hash_1": history["root_history"][1]["root_hash"].hex(),
"hash_2": history["root_history"][2]["root_hash"].hex(),
}
)
assert len(diff_res["diff"]) == 3
diff1 = {"type": "DELETE", "key": key1.hex(), "value": value1.hex()}
diff4 = {"type": "INSERT", "key": key4.hex(), "value": value4.hex()}
diff5 = {"type": "INSERT", "key": key5.hex(), "value": value5.hex()}
assert diff4 in diff_res["diff"]
assert diff5 in diff_res["diff"]
assert diff1 in diff_res["diff"]
@pytest.mark.asyncio
tests/core/data_layer/test_data_store.py
+55 -1
View File
@@ -23,6 +23,7 @@ from chia.data_layer.data_layer_types import (
DeletionData,
OperationType,
DiffData,
DownloadMode,
)
from chia.data_layer.data_layer_util import _debug_dump
from chia.data_layer.data_store import DataStore
@@ -30,7 +31,8 @@ from chia.types.blockchain_format.program import Program
from chia.types.blockchain_format.tree_hash import bytes32
from chia.util.byte_types import hexstr_to_bytes
from chia.util.db_wrapper import DBWrapper
from chia.data_layer.data_layer_types import Subscription
from chia.util.ints import uint16
from tests.core.data_layer.util import (
add_0123_example,
add_01234567_example,
@@ -1111,5 +1113,57 @@ async def test_kv_diff(data_store: DataStore, tree_id: bytes32) -> None:
root_start = await data_store.get_tree_root(tree_id)
root_end = await data_store.get_tree_root(tree_id)
assert root_start.node_hash is not None
assert root_end.node_hash is not None
diffs = await data_store.get_kv_diff(tree_id, root_start.node_hash, root_end.node_hash)
assert diffs == expected_diff
@pytest.mark.asyncio
async def test_kv_diff_2(data_store: DataStore, tree_id: bytes32) -> None:
node_hash = await data_store.insert(
key=b"000",
value=b"000",
tree_id=tree_id,
reference_node_hash=None,
side=None,
)
empty_hash = bytes32([0] * 32)
invalid_hash = bytes32([0] * 31 + [1])
diff_1 = await data_store.get_kv_diff(tree_id, empty_hash, node_hash)
assert diff_1 == set([DiffData(OperationType.INSERT, b"000", b"000")])
diff_2 = await data_store.get_kv_diff(tree_id, node_hash, empty_hash)
assert diff_2 == set([DiffData(OperationType.DELETE, b"000", b"000")])
diff_3 = await data_store.get_kv_diff(tree_id, invalid_hash, node_hash)
assert diff_3 == set()
@pytest.mark.asyncio
async def test_rollback_to_generation(data_store: DataStore, tree_id: bytes32) -> None:
await add_0123_example(data_store, tree_id)
expected_hashes = []
roots = await data_store.get_roots_between(tree_id, 1, 5)
for generation, root in enumerate(roots):
expected_hashes.append((generation + 1, root.node_hash))
for generation, expected_hash in reversed(expected_hashes):
await data_store.rollback_to_generation(tree_id, generation)
root = await data_store.get_tree_root(tree_id)
assert root.node_hash == expected_hash
@pytest.mark.asyncio
async def test_subscribe_unsubscribe(data_store: DataStore, tree_id: bytes32) -> None:
await data_store.subscribe(Subscription(tree_id, DownloadMode.HISTORY, "127.0.0.1", uint16(8000)))
assert await data_store.get_subscriptions() == [
Subscription(tree_id, DownloadMode.HISTORY, "127.0.0.1", uint16(8000))
]
await data_store.update_existing_subscription(Subscription(tree_id, DownloadMode.HISTORY, "0.0.0.0", uint16(8000)))
assert await data_store.get_subscriptions() == [
Subscription(tree_id, DownloadMode.HISTORY, "0.0.0.0", uint16(8000))
]
await data_store.update_existing_subscription(Subscription(tree_id, DownloadMode.HISTORY, "0.0.0.0", uint16(8001)))
assert await data_store.get_subscriptions() == [
Subscription(tree_id, DownloadMode.HISTORY, "0.0.0.0", uint16(8001))
]
await data_store.unsubscribe(tree_id)
assert await data_store.get_subscriptions() == []
tests/wallet/db_wallet/test_dl_wallet.py
+184 -24
View File
@@ -4,7 +4,8 @@ from typing import AsyncIterator, Iterator
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.types.peer_info import PeerInfo
from chia.util.ints import uint16, uint64
from chia.util.ints import uint16, uint32, uint64
from chia.simulator.simulator_protocol import FarmNewBlockProtocol
from tests.setup_nodes import setup_simulators_and_wallets
from chia.data_layer.data_layer_wallet import DataLayerWallet
@@ -24,6 +25,16 @@ def event_loop() -> Iterator[asyncio.AbstractEventLoop]:
yield loop
async def is_singleton_confirmed(dl_wallet: DataLayerWallet, lid: bytes32) -> bool:
rec = await dl_wallet.get_latest_singleton(lid)
if rec is None:
return False
if rec.confirmed is True:
assert rec.confirmed_at_height > 0
assert rec.timestamp > 0
return rec.confirmed
class TestDLWallet:
@pytest.fixture(scope="function")
async def wallet_node(self) -> AsyncIterator[SimulatorsAndWallets]:
@@ -82,12 +93,6 @@ class TestDLWallet:
current_tree = MerkleTree(nodes)
current_root = current_tree.calculate_root()
async def is_singleton_confirmed(lid: bytes32) -> bool:
rec = await dl_wallet.get_latest_singleton(lid)
if rec is None:
return False
return rec.confirmed
for i in range(0, 2):
dl_record, std_record, launcher_id = await dl_wallet.generate_new_reporter(
current_root, fee=uint64(1999999999999)
@@ -99,7 +104,7 @@ class TestDLWallet:
await wallet_node_0.wallet_state_manager.add_pending_transaction(std_record)
await full_node_api.process_transaction_records(records=[dl_record, std_record])
await time_out_assert(15, is_singleton_confirmed, True, launcher_id)
await time_out_assert(15, is_singleton_confirmed, True, dl_wallet, launcher_id)
await asyncio.sleep(0.5)
await time_out_assert(10, wallet_0.get_unconfirmed_balance, 0)
@@ -146,12 +151,6 @@ class TestDLWallet:
current_tree = MerkleTree(nodes)
current_root = current_tree.calculate_root()
async def is_singleton_confirmed(wallet: DataLayerWallet, lid: bytes32) -> bool:
latest_singleton = await wallet.get_latest_singleton(lid)
if latest_singleton is None:
return False
return latest_singleton.confirmed
dl_record, std_record, launcher_id = await dl_wallet_0.generate_new_reporter(current_root)
assert await dl_wallet_0.get_latest_singleton(launcher_id) is not None
@@ -219,12 +218,6 @@ class TestDLWallet:
current_tree = MerkleTree(nodes)
current_root = current_tree.calculate_root()
async def is_singleton_confirmed(lid: bytes32) -> bool:
latest_singleton = await dl_wallet.get_latest_singleton(lid)
if latest_singleton is None:
return False
return latest_singleton.confirmed
dl_record, std_record, launcher_id = await dl_wallet.generate_new_reporter(current_root)
assert await dl_wallet.get_latest_singleton(launcher_id) is not None
@@ -233,7 +226,7 @@ class TestDLWallet:
await wallet_node_0.wallet_state_manager.add_pending_transaction(std_record)
await full_node_api.process_transaction_records(records=[dl_record, std_record])
await time_out_assert(15, is_singleton_confirmed, True, launcher_id)
await time_out_assert(15, is_singleton_confirmed, True, dl_wallet, launcher_id)
await asyncio.sleep(0.5)
previous_record = await dl_wallet.get_latest_singleton(launcher_id)
@@ -253,7 +246,7 @@ class TestDLWallet:
await wallet_node_0.wallet_state_manager.add_pending_transaction(tx)
await full_node_api.process_transaction_records(records=txs)
await time_out_assert(15, is_singleton_confirmed, True, launcher_id)
await time_out_assert(15, is_singleton_confirmed, True, dl_wallet, launcher_id)
await time_out_assert(10, wallet_0.get_unconfirmed_balance, funds - 2000000000000)
await time_out_assert(10, wallet_0.get_confirmed_balance, funds - 2000000000000)
await asyncio.sleep(0.5)
@@ -270,7 +263,7 @@ class TestDLWallet:
await wallet_node_0.wallet_state_manager.add_pending_transaction(tx)
await full_node_api.process_transaction_records(records=txs)
await time_out_assert(15, is_singleton_confirmed, True, launcher_id)
await time_out_assert(15, is_singleton_confirmed, True, dl_wallet, launcher_id)
await asyncio.sleep(0.5)
await time_out_assert(10, wallet_0.get_unconfirmed_balance, funds - 6000000000000)
@@ -289,9 +282,176 @@ class TestDLWallet:
await wallet_node_0.wallet_state_manager.add_pending_transaction(tx)
await full_node_api.process_transaction_records(records=txs)
await time_out_assert(15, is_singleton_confirmed, True, launcher_id)
await time_out_assert(15, is_singleton_confirmed, True, dl_wallet, launcher_id)
await asyncio.sleep(0.5)
@pytest.mark.parametrize(
"trusted",
[True, False],
)
@pytest.mark.asyncio
async def test_rebase(self, two_wallet_nodes: SimulatorsAndWallets, trusted: bool) -> None:
full_nodes, wallets = two_wallet_nodes
full_node_api = full_nodes[0]
full_node_server = full_node_api.server
wallet_node_0, server_0 = wallets[0]
wallet_node_1, server_1 = wallets[1]
assert wallet_node_0.wallet_state_manager is not None
assert wallet_node_1.wallet_state_manager is not None
wallet_0 = wallet_node_0.wallet_state_manager.main_wallet
wallet_1 = wallet_node_1.wallet_state_manager.main_wallet
if trusted:
wallet_node_0.config["trusted_peers"] = {full_node_server.node_id.hex(): full_node_server.node_id.hex()}
wallet_node_1.config["trusted_peers"] = {full_node_server.node_id.hex(): full_node_server.node_id.hex()}
else:
wallet_node_0.config["trusted_peers"] = {}
wallet_node_1.config["trusted_peers"] = {}
await server_0.start_client(PeerInfo("localhost", uint16(full_node_server._port)), None)
await server_1.start_client(PeerInfo("localhost", uint16(full_node_server._port)), None)
funds = await full_node_api.farm_blocks(count=5, wallet=wallet_0)
await full_node_api.farm_blocks(count=5, wallet=wallet_1)
await time_out_assert(10, wallet_0.get_unconfirmed_balance, funds)
await time_out_assert(10, wallet_0.get_confirmed_balance, funds)
await time_out_assert(10, wallet_1.get_unconfirmed_balance, funds)
await time_out_assert(10, wallet_1.get_confirmed_balance, funds)
async with wallet_node_0.wallet_state_manager.lock:
dl_wallet_0 = await DataLayerWallet.create_new_dl_wallet(wallet_node_0.wallet_state_manager, wallet_0)
async with wallet_node_1.wallet_state_manager.lock:
dl_wallet_1 = await DataLayerWallet.create_new_dl_wallet(wallet_node_1.wallet_state_manager, wallet_1)
nodes = [Program.to("thing").get_tree_hash(), Program.to([8]).get_tree_hash()]
current_tree = MerkleTree(nodes)
current_root = current_tree.calculate_root()
async def is_singleton_confirmed(wallet: DataLayerWallet, lid: bytes32) -> bool:
latest_singleton = await wallet.get_latest_singleton(lid)
if latest_singleton is None:
return False
return latest_singleton.confirmed
dl_record, std_record, launcher_id = await dl_wallet_0.generate_new_reporter(current_root)
initial_record = await dl_wallet_0.get_latest_singleton(launcher_id)
assert initial_record is not None
await wallet_node_0.wallet_state_manager.add_pending_transaction(dl_record)
await wallet_node_0.wallet_state_manager.add_pending_transaction(std_record)
await full_node_api.process_transaction_records(records=[dl_record, std_record])
await time_out_assert(15, is_singleton_confirmed, True, dl_wallet_0, launcher_id)
await asyncio.sleep(0.5)
await dl_wallet_1.track_new_launcher_id(launcher_id)
await time_out_assert(15, is_singleton_confirmed, True, dl_wallet_1, launcher_id)
current_record = await dl_wallet_1.get_latest_singleton(launcher_id)
assert current_record is not None
await asyncio.sleep(0.5)
# Because these have the same fee, the one that gets pushed first will win
report_txs, _ = await dl_wallet_1.create_report_spend(launcher_id, fee=uint64(2000000000000))
record_1 = await dl_wallet_1.get_latest_singleton(launcher_id)
assert record_1 is not None
assert current_record != record_1
update_txs = await dl_wallet_0.create_update_state_spend(
launcher_id, bytes32([0] * 32), fee=uint64(2000000000000)
)
record_0 = await dl_wallet_0.get_latest_singleton(launcher_id)
assert record_0 is not None
assert initial_record != record_0
assert record_0 != record_1
for tx in report_txs:
await wallet_node_1.wallet_state_manager.add_pending_transaction(tx)
for tx in update_txs:
await wallet_node_0.wallet_state_manager.add_pending_transaction(tx)
for i in range(0, 2):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(bytes32(32 * b"0")))
await asyncio.sleep(0.5)
async def is_singleton_generation(wallet: DataLayerWallet, launcher_id: bytes32, generation: int) -> bool:
latest = await wallet.get_latest_singleton(launcher_id)
if latest is not None and latest.generation == generation:
return True
return False
next_generation = current_record.generation + 2
await time_out_assert(15, is_singleton_generation, True, dl_wallet_0, launcher_id, next_generation)
for i in range(0, 2):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(bytes32(32 * b"0")))
await asyncio.sleep(0.5)
await time_out_assert(15, is_singleton_confirmed, True, dl_wallet_0, launcher_id)
await time_out_assert(15, is_singleton_generation, True, dl_wallet_1, launcher_id, next_generation)
latest = await dl_wallet_0.get_latest_singleton(launcher_id)
assert latest is not None
assert latest == (await dl_wallet_1.get_latest_singleton(launcher_id))
await time_out_assert(15, wallet_0.get_confirmed_balance, funds - 2000000000001)
await time_out_assert(15, wallet_0.get_unconfirmed_balance, funds - 2000000000001)
assert (
len(
await dl_wallet_0.get_history(
launcher_id, min_generation=uint32(next_generation - 1), max_generation=uint32(next_generation - 1)
)
)
== 1
)
for tx in update_txs:
assert await wallet_node_0.wallet_state_manager.tx_store.get_transaction_record(tx.name) is None
assert await dl_wallet_0.get_singleton_record(record_0.coin_id) is None
update_txs_1 = await dl_wallet_0.create_update_state_spend(launcher_id, bytes32([1] * 32))
record_1 = await dl_wallet_0.get_latest_singleton(launcher_id)
assert record_1 is not None
for tx in update_txs_1:
await wallet_node_0.wallet_state_manager.add_pending_transaction(tx)
await full_node_api.wait_transaction_records_entered_mempool(update_txs_1)
# Delete any trace of that update
await wallet_node_0.wallet_state_manager.dl_store.delete_singleton_record(record_1.coin_id)
for tx in update_txs_1:
await wallet_node_0.wallet_state_manager.tx_store.delete_transaction_record(tx.name)
update_txs_0 = await dl_wallet_0.create_update_state_spend(
launcher_id, bytes32([2] * 32), fee=uint64(2000000000000)
)
record_0 = await dl_wallet_0.get_latest_singleton(launcher_id)
assert record_0 is not None
assert record_0 != record_1
for tx in update_txs_0:
await wallet_node_0.wallet_state_manager.add_pending_transaction(tx)
for i in range(0, 2):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(bytes32(32 * b"0")))
await asyncio.sleep(0.5)
next_generation += 1
await time_out_assert(15, is_singleton_generation, True, dl_wallet_0, launcher_id, next_generation)
latest = await dl_wallet_0.get_latest_singleton(launcher_id)
assert latest is not None
assert latest.root == bytes32([1] * 32)
await time_out_assert(15, wallet_0.get_confirmed_balance, funds - 2000000000001)
await time_out_assert(15, wallet_0.get_unconfirmed_balance, funds - 2000000000001)
assert (
len(
await dl_wallet_0.get_history(
launcher_id, min_generation=uint32(next_generation), max_generation=uint32(next_generation)
)
)
== 1
)
for tx in update_txs_0:
assert await wallet_node_0.wallet_state_manager.tx_store.get_transaction_record(tx.name) is None
assert await dl_wallet_0.get_singleton_record(record_0.coin_id) is None
# @pytest.mark.skip(reason="DLO Wallet is not supported yet")
# @pytest.mark.asyncio
# async def test_dlo_wallet(self, three_wallet_nodes: SimulatorsAndWallets) -> None: