from __future__ import annotations import asyncio import contextlib import dataclasses import functools import json import logging import os import random import time import traceback from collections.abc import AsyncIterator, Awaitable from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar, cast, final import aiohttp from chia_rs.datalayer import ProofOfInclusion, ProofOfInclusionLayer from chia_rs.sized_bytes import bytes32 from chia_rs.sized_ints import uint32, uint64 from chia.data_layer.data_layer_errors import KeyNotFoundError from chia.data_layer.data_layer_util import ( DiffData, InternalNode, KeysPaginationData, KeysValuesPaginationData, KeyValue, KVDiffPaginationData, Layer, Offer, OfferStore, PluginRemote, PluginStatus, Proof, Root, ServerInfo, Side, Status, StoreProofs, Subscription, SyncStatus, TerminalNode, Unspecified, UnsubscribeData, calculate_sibling_sides_integer, collect_sibling_hashes, get_delta_filename_path, get_full_tree_filename_path, leaf_hash, unspecified, ) from chia.data_layer.data_layer_wallet import DataLayerWallet, Mirror, verify_offer from chia.data_layer.data_store import DataStore from chia.data_layer.download_data import delete_full_file_if_exists, insert_from_delta_file, write_files_for_root from chia.data_layer.singleton_record import SingletonRecord from chia.protocols.outbound_message import NodeType from chia.rpc.rpc_server import StateChangedProtocol, default_get_connections from chia.server.server import ChiaServer from chia.server.ws_connection import WSChiaConnection from chia.util.async_pool import Job, QueuedAsyncPool from chia.util.path import path_from_root from chia.util.task_referencer import create_referenced_task from chia.wallet.puzzle_drivers import Solver from chia.wallet.trade_record import TradeRecord from chia.wallet.trading.offer import Offer as TradingOffer from chia.wallet.transaction_record import TransactionRecord from chia.wallet.util.tx_config import DEFAULT_TX_CONFIG from chia.wallet.wallet_request_types import ( CancelOffer, CreateNewDL, CreateOfferForIDs, DLDeleteMirror, DLGetMirrors, DLHistory, DLLatestSingleton, DLNewMirror, DLStopTracking, DLTrackNew, DLUpdateMultiple, DLUpdateMultipleUpdates, DLUpdateRoot, GetOffer, LauncherRootPair, LogIn, TakeOffer, ) from chia.wallet.wallet_rpc_client import WalletRpcClient def server_files_path_from_config(config: dict[str, Any], root_path: Path) -> Path: server_files_replaced: Path = path_from_root( root_path, config.get("server_files_location", "data_layer/db/server_files_location_CHALLENGE").replace( "CHALLENGE", config["selected_network"] ), ) return server_files_replaced async def get_plugin_info(plugin_remote: PluginRemote) -> tuple[PluginRemote, dict[str, Any]]: try: async with aiohttp.ClientSession() as session: async with session.post( plugin_remote.url + "/plugin_info", json={}, headers=plugin_remote.headers, ) as response: ret = {"status": response.status} if response.status == 200: ret["response"] = json.loads(await response.text()) return plugin_remote, ret except aiohttp.ClientError as e: return plugin_remote, {"error": f"ClientError: {e}"} @final @dataclasses.dataclass class DataLayer: if TYPE_CHECKING: from chia.rpc.rpc_server import RpcServiceProtocol _protocol_check: ClassVar[RpcServiceProtocol] = cast("DataLayer", None) db_path: Path merkle_blobs_path: Path key_value_blobs_path: Path config: dict[str, Any] root_path: Path log: logging.Logger wallet_rpc_init: Awaitable[WalletRpcClient] downloaders: list[PluginRemote] uploaders: list[PluginRemote] maximum_full_file_count: int server_files_location: Path unsubscribe_data_queue: list[UnsubscribeData] _server: ChiaServer | None = None none_bytes: bytes32 = bytes32.zeros initialized: bool = False _data_store: DataStore | None = None state_changed_callback: StateChangedProtocol | None = None _shut_down: bool = False periodically_manage_data_task: asyncio.Task[None] | None = None _wallet_rpc: WalletRpcClient | None = None subscription_lock: asyncio.Lock = dataclasses.field(default_factory=asyncio.Lock) subscription_update_concurrency: int = 5 client_timeout: aiohttp.ClientTimeout = dataclasses.field( default_factory=functools.partial(aiohttp.ClientTimeout, total=45, sock_connect=5) ) group_files_by_store: bool = False @property def server(self) -> ChiaServer: # This is a stop gap until the class usage is refactored such the values of # integral attributes are known at creation of the instance. if self._server is None: raise RuntimeError("server not assigned") return self._server @property def data_store(self) -> DataStore: # This is a stop gap until the class usage is refactored such the values of # integral attributes are known at creation of the instance. if self._data_store is None: raise RuntimeError("data_store not assigned") return self._data_store @property def wallet_rpc(self) -> WalletRpcClient: # This is a stop gap until the class usage is refactored such the values of # integral attributes are known at creation of the instance. if self._wallet_rpc is None: raise RuntimeError("wallet_rpc not assigned") return self._wallet_rpc @classmethod def create( cls, config: dict[str, Any], root_path: Path, wallet_rpc_init: Awaitable[WalletRpcClient], downloaders: list[PluginRemote], uploaders: list[PluginRemote], # dont add FilesystemUploader to this, it is the default uploader name: str | None = None, ) -> DataLayer: if name == "": # TODO: If no code depends on "" counting as 'unspecified' then we do not # need this. name = None server_files_replaced = server_files_path_from_config(config, root_path) db_path_replaced: str = config["database_path"].replace("CHALLENGE", config["selected_network"]) merkle_blobs_path_replaced: str = config.get( "merkle_blobs_path", "data_layer/db/merkle_blobs_CHALLENGE" ).replace("CHALLENGE", config["selected_network"]) key_value_blobs_path_replaced: str = config.get( "key_value_blobs_path", "data_layer/db/key_value_blobs_CHALLENGE" ).replace("CHALLENGE", config["selected_network"]) self = cls( config=config, root_path=root_path, wallet_rpc_init=wallet_rpc_init, log=logging.getLogger(name if name is None else __name__), db_path=path_from_root(root_path, db_path_replaced), merkle_blobs_path=path_from_root(root_path, merkle_blobs_path_replaced), key_value_blobs_path=path_from_root(root_path, key_value_blobs_path_replaced), server_files_location=server_files_replaced, downloaders=downloaders, uploaders=uploaders, maximum_full_file_count=config.get("maximum_full_file_count", 1), subscription_update_concurrency=config.get("subscription_update_concurrency", 5), unsubscribe_data_queue=[], client_timeout=aiohttp.ClientTimeout( total=config.get("client_timeout", 45), sock_connect=config.get("connect_timeout", 5) ), group_files_by_store=config.get("group_files_by_store", False), ) self.db_path.parent.mkdir(parents=True, exist_ok=True) self.merkle_blobs_path.mkdir(parents=True, exist_ok=True) self.key_value_blobs_path.mkdir(parents=True, exist_ok=True) self.server_files_location.mkdir(parents=True, exist_ok=True) return self @contextlib.asynccontextmanager async def manage(self) -> AsyncIterator[None]: sql_log_path: Path | None = None if self.config.get("log_sqlite_cmds", False): sql_log_path = path_from_root(self.root_path, "log/data_sql.log") self.log.info(f"logging SQL commands to {sql_log_path}") cache_capacity = self.config.get("merkle_blobs_cache_size", 1) async with DataStore.managed( database=self.db_path, merkle_blobs_path=self.merkle_blobs_path, key_value_blobs_path=self.key_value_blobs_path, sql_log_path=sql_log_path, cache_capacity=cache_capacity, ) as self._data_store: self._wallet_rpc = await self.wallet_rpc_init await self._data_store.migrate_db(self.server_files_location) self.periodically_manage_data_task = create_referenced_task(self.periodically_manage_data()) try: yield finally: # TODO: review for anything else we need to do here self._shut_down = True if self._wallet_rpc is not None: self.wallet_rpc.close() if self.periodically_manage_data_task is not None: try: self.periodically_manage_data_task.cancel() except asyncio.CancelledError: pass if self._wallet_rpc is not None: await self.wallet_rpc.await_closed() def _set_state_changed_callback(self, callback: StateChangedProtocol) -> None: self.state_changed_callback = callback async def on_connect(self, connection: WSChiaConnection) -> None: pass def get_connections(self, request_node_type: NodeType | None) -> list[dict[str, Any]]: return default_get_connections(server=self.server, request_node_type=request_node_type) def set_server(self, server: ChiaServer) -> None: self._server = server async def wallet_log_in(self, fingerprint: int) -> int: try: result = await self.wallet_rpc.log_in(LogIn(fingerprint=uint32(fingerprint))) except ValueError as e: raise Exception(f"DataLayer wallet RPC log in request failed: {e.args[0]}") return result.fingerprint async def create_store(self, fee: uint64, root: bytes32 = bytes32.zeros) -> tuple[list[TransactionRecord], bytes32]: create_res = await self.wallet_rpc.create_new_dl(CreateNewDL(root=root, fee=fee, push=True), DEFAULT_TX_CONFIG) res = await self.data_store.create_tree(store_id=create_res.launcher_id) if res is None: self.log.fatal("failed creating store") self.initialized = True return create_res.transactions, create_res.launcher_id async def batch_update( self, store_id: bytes32, changelist: list[dict[str, Any]], fee: uint64, submit_on_chain: bool = True, ) -> TransactionRecord | None: status = Status.PENDING if submit_on_chain else Status.PENDING_BATCH await self.batch_insert(store_id=store_id, changelist=changelist, status=status) if submit_on_chain: return await self.publish_update(store_id=store_id, fee=fee) else: return None async def _get_publishable_root_hash(self, store_id: bytes32) -> bytes32: pending_root: Root | None = await self.data_store.get_pending_root(store_id=store_id) if pending_root is None: raise Exception("Latest root is already confirmed.") if pending_root.status == Status.PENDING_BATCH: raise Exception("Unable to publish on chain, batch update set still open.") return self.none_bytes if pending_root.node_hash is None else pending_root.node_hash async def multistore_batch_update( self, store_updates: list[dict[str, Any]], fee: uint64, submit_on_chain: bool = True, ) -> list[TransactionRecord]: store_ids: set[bytes32] = set() for update in store_updates: store_id = update["store_id"] changelist = update["changelist"] if store_id in store_ids: raise Exception(f"Store id {store_id.hex()} must appear in a single update") store_ids.add(store_id) status = Status.PENDING if submit_on_chain else Status.PENDING_BATCH await self.batch_insert(store_id=store_id, changelist=changelist, status=status) if submit_on_chain: updates: list[LauncherRootPair] = [] for store_id in store_ids: await self._update_confirmation_status(store_id=store_id) root_hash = await self._get_publishable_root_hash(store_id=store_id) updates.append(LauncherRootPair(launcher_id=store_id, new_root=root_hash)) response = await self.wallet_rpc.dl_update_multiple( DLUpdateMultiple(updates=DLUpdateMultipleUpdates(launcher_root_pairs=updates), fee=fee), DEFAULT_TX_CONFIG, ) return response.transactions else: return [] async def submit_pending_root( self, store_id: bytes32, fee: uint64, ) -> TransactionRecord: await self._update_confirmation_status(store_id=store_id) pending_root: Root | None = await self.data_store.get_pending_root(store_id=store_id) if pending_root is None: raise Exception("Latest root is already confirmed.") if pending_root.status == Status.PENDING: raise Exception("Pending root is already submitted.") await self.data_store.change_root_status(pending_root, Status.PENDING) return await self.publish_update(store_id, fee) async def submit_all_pending_roots(self, fee: uint64) -> list[TransactionRecord]: pending_roots = await self.data_store.get_all_pending_batches_roots() updates: list[LauncherRootPair] = [] if len(pending_roots) == 0: raise Exception("No pending roots found to submit") for pending_root in pending_roots: root_hash = pending_root.node_hash if pending_root.node_hash is not None else self.none_bytes updates.append(LauncherRootPair(launcher_id=pending_root.store_id, new_root=root_hash)) await self.data_store.change_root_status(pending_root, Status.PENDING) response = await self.wallet_rpc.dl_update_multiple( DLUpdateMultiple(updates=DLUpdateMultipleUpdates(launcher_root_pairs=updates), fee=fee), DEFAULT_TX_CONFIG, ) return response.transactions async def batch_insert( self, store_id: bytes32, changelist: list[dict[str, Any]], status: Status = Status.PENDING, enable_batch_autoinsert: bool | None = None, ) -> bytes32: await self._update_confirmation_status(store_id=store_id) async with self.data_store.transaction(): pending_root: Root | None = await self.data_store.get_pending_root(store_id=store_id) if pending_root is not None and pending_root.status == Status.PENDING: raise Exception("Already have a pending root waiting for confirmation.") # check before any DL changes that this singleton is currently owned by this wallet singleton_records: list[SingletonRecord] = await self.get_owned_stores() if not any(store_id == singleton.launcher_id for singleton in singleton_records): raise ValueError(f"Singleton with launcher ID {store_id} is not owned by DL Wallet") t1 = time.monotonic() if enable_batch_autoinsert is None: enable_batch_autoinsert = self.config.get("enable_batch_autoinsert", True) batch_hash = await self.data_store.insert_batch(store_id, changelist, status, enable_batch_autoinsert) t2 = time.monotonic() self.log.info(f"Data store batch update process time: {t2 - t1}.") # todo return empty node hash from get_tree_root if batch_hash is not None: node_hash = batch_hash else: node_hash = self.none_bytes # todo change return node_hash async def publish_update( self, store_id: bytes32, fee: uint64, ) -> TransactionRecord: await self._update_confirmation_status(store_id=store_id) root_hash = await self._get_publishable_root_hash(store_id=store_id) transaction_record = ( await self.wallet_rpc.dl_update_root( DLUpdateRoot( launcher_id=store_id, new_root=root_hash, fee=fee, push=True, ), DEFAULT_TX_CONFIG, ) ).tx_record return transaction_record async def get_key_value_hash( self, store_id: bytes32, key: bytes, root_hash: bytes32 | Unspecified = unspecified, ) -> bytes32: await self._update_confirmation_status(store_id=store_id) async with self.data_store.transaction(): node = await self.data_store.get_node_by_key(store_id=store_id, key=key, root_hash=root_hash) return node.hash async def get_value(self, store_id: bytes32, key: bytes, root_hash: bytes32 | Unspecified = unspecified) -> bytes: await self._update_confirmation_status(store_id=store_id) async with self.data_store.transaction(): # this either returns the node or raises an exception res = await self.data_store.get_node_by_key(store_id=store_id, key=key, root_hash=root_hash) return res.value async def get_keys_values( self, store_id: bytes32, root_hash: bytes32 | Unspecified, ) -> list[TerminalNode]: await self._update_confirmation_status(store_id=store_id) res = await self.data_store.get_keys_values(store_id, root_hash) if res is None: self.log.error("Failed to fetch keys values") return res async def get_keys_values_paginated( self, store_id: bytes32, root_hash: bytes32 | Unspecified, page: int, max_page_size: int | None = None, ) -> KeysValuesPaginationData: await self._update_confirmation_status(store_id=store_id) if max_page_size is None: max_page_size = 40 * 1024 * 1024 res = await self.data_store.get_keys_values_paginated(store_id, page, max_page_size, root_hash) return res async def get_keys(self, store_id: bytes32, root_hash: bytes32 | Unspecified) -> list[bytes]: await self._update_confirmation_status(store_id=store_id) res = await self.data_store.get_keys(store_id, root_hash) return res async def get_keys_paginated( self, store_id: bytes32, root_hash: bytes32 | Unspecified, page: int, max_page_size: int | None = None, ) -> KeysPaginationData: await self._update_confirmation_status(store_id=store_id) if max_page_size is None: max_page_size = 40 * 1024 * 1024 res = await self.data_store.get_keys_paginated(store_id, page, max_page_size, root_hash) return res async def get_ancestors(self, node_hash: bytes32, store_id: bytes32) -> list[InternalNode]: await self._update_confirmation_status(store_id=store_id) res = await self.data_store.get_ancestors(node_hash=node_hash, store_id=store_id) if res is None: self.log.error("Failed to get ancestors") return res async def get_root(self, store_id: bytes32) -> SingletonRecord | None: latest = ( await self.wallet_rpc.dl_latest_singleton(DLLatestSingleton(launcher_id=store_id, only_confirmed=True)) ).singleton 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) -> bytes32 | None: await self._update_confirmation_status(store_id=store_id) res = await self.data_store.get_tree_root(store_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(DLHistory(launcher_id=store_id))).history if records is None: self.log.error(f"Failed to get root history for {store_id.hex()}") root_history = [] prev: SingletonRecord | None = 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 _update_confirmation_status(self, store_id: bytes32) -> None: async with self.data_store.transaction(): try: root = await self.data_store.get_tree_root(store_id=store_id) except Exception: root = None singleton_record = ( await self.wallet_rpc.dl_latest_singleton(DLLatestSingleton(launcher_id=store_id, only_confirmed=True)) ).singleton if singleton_record is None: return if root is None: pending_root = await self.data_store.get_pending_root(store_id=store_id) if pending_root is not None and pending_root.status == Status.PENDING: if pending_root.generation == 0 and pending_root.node_hash is None: await self.data_store.change_root_status(pending_root, Status.COMMITTED) await self.data_store.clear_pending_roots(store_id=store_id) return else: root = None if root is None: self.log.info(f"Don't have pending root for {store_id}.") return if root.generation == singleton_record.generation: return if root.generation > singleton_record.generation: self.log.info( f"Local root ahead of chain root: {root.generation} {singleton_record.generation}. " "Maybe we're doing a batch update." ) return wallet_history = ( await self.wallet_rpc.dl_history( DLHistory( launcher_id=store_id, min_generation=uint32(root.generation + 1), max_generation=singleton_record.generation, ) ) ).history new_hashes = [record.root for record in reversed(wallet_history)] root_hash = self.none_bytes if root.node_hash is None else root.node_hash generation_shift = 0 while len(new_hashes) > 0 and new_hashes[0] == root_hash: generation_shift += 1 new_hashes.pop(0) if generation_shift > 0: await self.data_store.clear_pending_roots(store_id=store_id) await self.data_store.shift_root_generations(store_id=store_id, shift_size=generation_shift) else: expected_root_hash = None if new_hashes[0] == self.none_bytes else new_hashes[0] pending_root = await self.data_store.get_pending_root(store_id=store_id) if ( pending_root is not None and pending_root.generation == root.generation + 1 and pending_root.node_hash == expected_root_hash and pending_root.status == Status.PENDING ): await self.data_store.change_root_status(pending_root, Status.COMMITTED) await self.data_store.clear_pending_roots(store_id=store_id) async def fetch_and_validate(self, store_id: bytes32) -> None: singleton_record = ( await self.wallet_rpc.dl_latest_singleton(DLLatestSingleton(launcher_id=store_id, only_confirmed=True)) ).singleton if singleton_record is None: self.log.info(f"Fetch data: No singleton record for {store_id}.") return if singleton_record.generation == uint32(0): self.log.info(f"Fetch data: No data on chain for {store_id}.") return await self._update_confirmation_status(store_id=store_id) if not await self.data_store.store_id_exists(store_id=store_id): await self.data_store.create_tree(store_id=store_id, status=Status.COMMITTED) timestamp = int(time.time()) servers_info = await self.data_store.get_available_servers_for_store(store_id, timestamp) # TODO: maybe append a random object to the whole DataLayer class? random.shuffle(servers_info) success = False for server_info in servers_info: url = server_info.url root = await self.data_store.get_tree_root(store_id=store_id) if root.generation > singleton_record.generation: self.log.info( "Fetch data: local DL store is ahead of chain generation. " f"Local root: {root}. Singleton: {singleton_record}" ) break if root.generation == singleton_record.generation: self.log.info(f"Fetch data: wallet generation matching on-chain generation: {store_id}.") break self.log.info( f"Downloading files {store_id}. " f"Current wallet generation: {root.generation}. " f"Target wallet generation: {singleton_record.generation}. " f"Server used: {url}." ) to_download = ( await self.wallet_rpc.dl_history( DLHistory( launcher_id=store_id, min_generation=uint32(root.generation + 1), max_generation=singleton_record.generation, ) ) ).history try: proxy_url = self.config.get("proxy_url", None) success = await insert_from_delta_file( self.data_store, store_id, root.generation, target_generation=singleton_record.generation, root_hashes=[record.root for record in reversed(to_download)], server_info=server_info, client_foldername=self.server_files_location, timeout=self.client_timeout, log=self.log, proxy_url=proxy_url, downloader=await self.get_downloader(store_id, url), group_files_by_store=self.group_files_by_store, maximum_full_file_count=self.maximum_full_file_count, ) if success: self.log.info( f"Finished downloading and validating {store_id}. " f"Wallet generation saved: {singleton_record.generation}. " f"Root hash saved: {singleton_record.root}." ) break except aiohttp.client_exceptions.ClientConnectorError: self.log.warning(f"Server {url} unavailable for {store_id}.") except Exception as e: self.log.warning(f"Exception while downloading files for {store_id}: {e} {traceback.format_exc()}.") # if there aren't any servers then don't try to write the full tree if not success and len(servers_info) > 0: root = await self.data_store.get_tree_root(store_id=store_id) if root.node_hash is None: return filename_full_tree = get_full_tree_filename_path( foldername=self.server_files_location, store_id=store_id, node_hash=root.node_hash, generation=root.generation, group_by_store=self.group_files_by_store, ) # Had trouble with this generation, so generate full file for the generation we currently have if not os.path.exists(filename_full_tree): with open(filename_full_tree, "wb") as writer: await self.data_store.write_tree_to_file( root=root, node_hash=root.node_hash, store_id=store_id, deltas_only=False, writer=writer, ) self.log.info(f"Successfully written full tree filename {filename_full_tree}.") async def get_downloader(self, store_id: bytes32, url: str) -> PluginRemote | None: request_json = {"store_id": store_id.hex(), "url": url} for d in self.downloaders: async with aiohttp.ClientSession() as session: try: async with session.post( d.url + "/handle_download", json=request_json, headers=d.headers, ) as response: res_json = await response.json() if res_json["handle_download"]: return d except Exception as e: self.log.error(f"get_downloader could not get response: {type(e).__name__}: {e}") return None async def clean_old_full_tree_files(self, store_id: bytes32) -> None: singleton_record = ( await self.wallet_rpc.dl_latest_singleton(DLLatestSingleton(launcher_id=store_id, only_confirmed=True)) ).singleton if singleton_record is None: return await self._update_confirmation_status(store_id=store_id) root = await self.data_store.get_tree_root(store_id=store_id) latest_generation = root.generation full_tree_first_publish_generation = max(0, latest_generation - self.maximum_full_file_count + 1) foldername = self.server_files_location for generation in range(full_tree_first_publish_generation - 1, 0, -1): root = await self.data_store.get_tree_root(store_id=store_id, generation=generation) file_exists = delete_full_file_if_exists(foldername, store_id, root) if not file_exists: break async def upload_files(self, store_id: bytes32) -> None: uploaders = await self.get_uploaders(store_id) singleton_record = ( await self.wallet_rpc.dl_latest_singleton(DLLatestSingleton(launcher_id=store_id, only_confirmed=True)) ).singleton if singleton_record is None: self.log.info(f"Upload files: no on-chain record for {store_id}.") return await self._update_confirmation_status(store_id=store_id) root = await self.data_store.get_tree_root(store_id=store_id) latest_generation = root.generation # Don't store full tree files before this generation. full_tree_first_publish_generation = max(0, latest_generation - self.maximum_full_file_count + 1) publish_generation = min(singleton_record.generation, 0 if root is None else root.generation) # If we make some batch updates, which get confirmed to the chain, we need to create the files. # We iterate back and write the missing files, until we find the files already written. root = await self.data_store.get_tree_root(store_id=store_id, generation=publish_generation) while publish_generation > 0: write_file_result = await write_files_for_root( self.data_store, store_id, root, self.server_files_location, full_tree_first_publish_generation, group_by_store=self.group_files_by_store, ) if not write_file_result.result: # this particular return only happens if the files already exist, no need to log anything break try: if uploaders is not None and len(uploaders) > 0: request_json = { "store_id": store_id.hex(), "diff_filename": write_file_result.diff_tree.name, "group_files_by_store": self.group_files_by_store, } if write_file_result.full_tree is not None: request_json["full_tree_filename"] = write_file_result.full_tree.name for uploader in uploaders: self.log.info(f"Using uploader {uploader} for store {store_id.hex()}") async with aiohttp.ClientSession() as session: async with session.post( uploader.url + "/upload", json=request_json, headers=uploader.headers, ) as response: res_json = await response.json() if res_json["uploaded"]: self.log.info( f"Uploaded files to {uploader} for store {store_id.hex()} " f"generation {publish_generation}" ) else: self.log.error( f"Failed to upload files to, will retry later: {uploader} : {res_json}" ) except Exception as e: self.log.error(f"Exception uploading files, will retry later: store id {store_id}") self.log.debug(f"Failed to upload files, cleaning local files: {type(e).__name__}: {e}") if write_file_result.full_tree is not None: os.remove(write_file_result.full_tree) os.remove(write_file_result.diff_tree) publish_generation -= 1 root = await self.data_store.get_tree_root(store_id=store_id, generation=publish_generation) async def add_missing_files(self, store_id: bytes32, overwrite: bool, foldername: Path | None) -> None: root = await self.data_store.get_tree_root(store_id=store_id) latest_generation = root.generation full_tree_first_publish_generation = max(0, latest_generation - self.maximum_full_file_count + 1) singleton_record = ( await self.wallet_rpc.dl_latest_singleton(DLLatestSingleton(launcher_id=store_id, only_confirmed=True)) ).singleton if singleton_record is None: self.log.error(f"No singleton record found for: {store_id}") return max_generation = min(singleton_record.generation, 0 if root is None else root.generation) server_files_location = foldername if foldername is not None else self.server_files_location files = [] for generation in range(1, max_generation + 1): root = await self.data_store.get_tree_root(store_id=store_id, generation=generation) res = await write_files_for_root( self.data_store, store_id, root, server_files_location, full_tree_first_publish_generation, overwrite, self.group_files_by_store, ) files.append(res.diff_tree.name) if res.full_tree is not None: files.append(res.full_tree.name) uploaders = await self.get_uploaders(store_id) if uploaders is not None and len(uploaders) > 0: request_json = { "store_id": store_id.hex(), "files": json.dumps(files), "group_files_by_store": self.group_files_by_store, } for uploader in uploaders: async with aiohttp.ClientSession() as session: async with session.post( uploader.url + "/add_missing_files", json=request_json, headers=uploader.headers, ) as response: res_json = await response.json() if not res_json["uploaded"]: self.log.error(f"failed to upload to uploader {uploader}") else: self.log.debug(f"uploaded to uploader {uploader}") async def subscribe(self, store_id: bytes32, urls: list[str]) -> Subscription: parsed_urls = [url.rstrip("/") for url in urls] subscription = Subscription(store_id, [ServerInfo(url, 0, 0) for url in parsed_urls]) await self.wallet_rpc.dl_track_new(DLTrackNew(launcher_id=subscription.store_id)) async with self.subscription_lock: await self.data_store.subscribe(subscription) self.log.info(f"Done adding subscription: {subscription.store_id}") return subscription async def remove_subscriptions(self, store_id: bytes32, urls: list[str]) -> None: parsed_urls = [url.rstrip("/") for url in urls] async with self.subscription_lock: await self.data_store.remove_subscriptions(store_id, parsed_urls) async def unsubscribe(self, store_id: bytes32, retain_data: bool) -> None: async with self.subscription_lock: subscriptions = await self.data_store.get_subscriptions() if store_id not in (subscription.store_id for subscription in subscriptions): raise RuntimeError("No subscription found for the given store_id.") # Unsubscribe is processed later, after all fetching of data is done, to avoid races. self.unsubscribe_data_queue.append(UnsubscribeData(store_id, retain_data)) async def process_unsubscribe(self, store_id: bytes32, retain_data: bool) -> None: # This function already acquired `subscriptions_lock`. subscriptions = await self.data_store.get_subscriptions() if store_id not in (subscription.store_id for subscription in subscriptions): raise RuntimeError("No subscription found for the given store_id.") paths: list[Path] = [] if await self.data_store.store_id_exists(store_id) and not retain_data: generation = await self.data_store.get_tree_generation(store_id) all_roots = await self.data_store.get_roots_between(store_id, 1, generation + 1) for root in all_roots: root_hash = root.node_hash if root.node_hash is not None else self.none_bytes for group_by_store in (True, False): paths.append( get_full_tree_filename_path( self.server_files_location, store_id, root_hash, root.generation, group_by_store, ) ) paths.append( get_delta_filename_path( self.server_files_location, store_id, root_hash, root.generation, group_by_store, ) ) # stop tracking first, then unsubscribe from the data store await self.wallet_rpc.dl_stop_tracking(DLStopTracking(launcher_id=store_id)) await self.data_store.unsubscribe(store_id) self.log.info(f"Unsubscribed to {store_id}") for file_path in paths: try: file_path.unlink() except FileNotFoundError: pass async def get_subscriptions(self) -> list[Subscription]: async with self.subscription_lock: return await self.data_store.get_subscriptions() async def add_mirror(self, store_id: bytes32, urls: list[str], amount: uint64, fee: uint64) -> None: if not urls: raise RuntimeError("URL list can't be empty") await self.wallet_rpc.dl_new_mirror( DLNewMirror(launcher_id=store_id, amount=amount, urls=urls, fee=fee, push=True), DEFAULT_TX_CONFIG ) async def delete_mirror(self, coin_id: bytes32, fee: uint64) -> None: await self.wallet_rpc.dl_delete_mirror(DLDeleteMirror(coin_id=coin_id, fee=fee, push=True), DEFAULT_TX_CONFIG) async def get_mirrors(self, store_id: bytes32) -> list[Mirror]: mirrors: list[Mirror] = (await self.wallet_rpc.dl_get_mirrors(DLGetMirrors(launcher_id=store_id))).mirrors return [mirror for mirror in mirrors if mirror.urls] async def update_subscriptions_from_wallet(self, store_id: bytes32) -> None: mirrors: list[Mirror] = (await self.wallet_rpc.dl_get_mirrors(DLGetMirrors(launcher_id=store_id))).mirrors urls: list[str] = [] for mirror in mirrors: urls += mirror.urls urls = [url.rstrip("/") for url in urls] await self.data_store.update_subscriptions_from_wallet(store_id, urls) async def get_owned_stores(self) -> list[SingletonRecord]: return (await self.wallet_rpc.dl_owned_singletons()).singletons async def get_kv_diff(self, store_id: bytes32, hash_1: bytes32, hash_2: bytes32) -> set[DiffData]: return await self.data_store.get_kv_diff(store_id, hash_1, hash_2) async def get_kv_diff_paginated( self, store_id: bytes32, # NOTE: empty is expressed as zeros hash_1: bytes32, hash_2: bytes32, page: int, max_page_size: int | None = None, ) -> KVDiffPaginationData: if max_page_size is None: max_page_size = 40 * 1024 * 1024 return await self.data_store.get_kv_diff_paginated(store_id, page, max_page_size, hash_1, hash_2) async def periodically_manage_data(self) -> None: manage_data_interval = self.config.get("manage_data_interval", 60) while not self._shut_down: async with self.subscription_lock: try: subscriptions = await self.data_store.get_subscriptions() for subscription in subscriptions: await self.wallet_rpc.dl_track_new(DLTrackNew(launcher_id=subscription.store_id)) break except aiohttp.client_exceptions.ClientConnectorError: pass except Exception as e: self.log.error(f"Exception while requesting wallet track subscription: {type(e)} {e}") self.log.warning("Cannot connect to the wallet. Retrying in 3s.") delay_until = time.monotonic() + 3 while time.monotonic() < delay_until: if self._shut_down: break await asyncio.sleep(0.1) while not self._shut_down: # Add existing subscriptions async with self.subscription_lock: subscriptions = await self.data_store.get_subscriptions() # pseudo-subscribe to all unsubscribed owned stores # Need this to make sure we process updates and generate DAT files try: owned_stores = await self.get_owned_stores() except (ValueError, aiohttp.client_exceptions.ClientConnectorError): # Sometimes the DL wallet isn't available, so we can't get the owned stores. # We'll try again next time. owned_stores = [] except Exception as e: self.log.error(f"Exception while fetching owned stores: {type(e)} {e} {traceback.format_exc()}") owned_stores = [] subscription_store_ids = {subscription.store_id for subscription in subscriptions} for record in owned_stores: store_id = record.launcher_id if store_id not in subscription_store_ids: try: # don't actually subscribe, just add to the list subscriptions.insert(0, Subscription(store_id=store_id, servers_info=[])) except Exception as e: self.log.info( f"Can't subscribe to owned store {store_id}: {type(e)} {e} {traceback.format_exc()}" ) # Optionally # Subscribe to all local non-owned store_ids that we can find on chain. # This is the prior behavior where all local stores, both owned and not owned, are subscribed to. if self.config.get("auto_subscribe_to_local_stores", False): local_store_ids = await self.data_store.get_store_ids() subscription_store_ids = {subscription.store_id for subscription in subscriptions} for local_id in local_store_ids: if local_id not in subscription_store_ids: try: subscription = await self.subscribe(local_id, []) subscriptions.insert(0, subscription) except Exception as e: self.log.info( f"Can't subscribe to local store {local_id}: {type(e)} {e} {traceback.format_exc()}" ) work_queue: asyncio.Queue[Job[Subscription]] = asyncio.Queue() async with QueuedAsyncPool.managed( name="DataLayer subscription update pool", worker_async_callable=self.update_subscription, job_queue=work_queue, target_worker_count=self.subscription_update_concurrency, log=self.log, ): jobs = [Job(input=subscription) for subscription in subscriptions] for job in jobs: await work_queue.put(job) await asyncio.gather(*(job.done.wait() for job in jobs), return_exceptions=True) # Do unsubscribes after the fetching of data is complete, to avoid races. async with self.subscription_lock: for unsubscribe_data in self.unsubscribe_data_queue: await self.process_unsubscribe(unsubscribe_data.store_id, unsubscribe_data.retain_data) self.unsubscribe_data_queue.clear() await asyncio.sleep(manage_data_interval) async def update_subscription( self, worker_id: int, job: Job[Subscription], ) -> None: subscription = job.input try: await self.update_subscriptions_from_wallet(subscription.store_id) await self.fetch_and_validate(subscription.store_id) await self.upload_files(subscription.store_id) await self.clean_old_full_tree_files(subscription.store_id) except Exception as e: self.log.error(f"Exception while fetching data: {type(e)} {e} {traceback.format_exc()}.") async def build_offer_changelist( self, store_id: bytes32, inclusions: tuple[KeyValue, ...], ) -> list[dict[str, Any]]: async with self.data_store.transaction(): changelist: list[dict[str, Any]] = [] for entry in inclusions: try: existing_value = await self.get_value(store_id=store_id, key=entry.key) except KeyNotFoundError: existing_value = None if existing_value == entry.value: # already present, nothing needed continue if existing_value is not None: # upsert, delete the existing key and value changelist.append( { "action": "delete", "key": entry.key, } ) changelist.append( { "action": "insert", "key": entry.key, "value": entry.value, } ) return changelist async def process_offered_stores(self, offer_stores: tuple[OfferStore, ...]) -> dict[bytes32, StoreProofs]: for offer_store in offer_stores: await self._update_confirmation_status(store_id=offer_store.store_id) async with self.data_store.transaction(): our_store_proofs: dict[bytes32, StoreProofs] = {} for offer_store in offer_stores: changelist = await self.build_offer_changelist( store_id=offer_store.store_id, inclusions=offer_store.inclusions, ) if len(changelist) > 0: new_root_hash = await self.batch_insert( store_id=offer_store.store_id, changelist=changelist, enable_batch_autoinsert=False, ) else: existing_root = await self.get_root(store_id=offer_store.store_id) if existing_root is None: raise Exception(f"store id not available: {offer_store.store_id.hex()}") new_root_hash = existing_root.root if new_root_hash is None: raise Exception("only inserts are supported so a None root hash should not be possible") proofs: list[Proof] = [] for entry in offer_store.inclusions: node_hash = await self.get_key_value_hash( store_id=offer_store.store_id, key=entry.key, root_hash=new_root_hash, ) proof_of_inclusion = await self.data_store.get_proof_of_inclusion_by_hash( node_hash=node_hash, store_id=offer_store.store_id, root_hash=new_root_hash, ) proof = Proof( key=entry.key, value=entry.value, node_hash=proof_of_inclusion.node_hash, layers=tuple( Layer( other_hash_side=Side(layer.other_hash_side), other_hash=layer.other_hash, combined_hash=layer.combined_hash, ) for layer in proof_of_inclusion.layers ), ) proofs.append(proof) store_proof = StoreProofs(store_id=offer_store.store_id, proofs=tuple(proofs)) our_store_proofs[offer_store.store_id] = store_proof return our_store_proofs async def make_offer( self, maker: tuple[OfferStore, ...], taker: tuple[OfferStore, ...], fee: uint64, ) -> Offer: async with self.data_store.transaction(): our_store_proofs = await self.process_offered_stores(offer_stores=maker) offer_dict: dict[str, str] = { **{offer_store.store_id.hex(): "-1" for offer_store in maker}, **{offer_store.store_id.hex(): "1" for offer_store in taker}, } solver = Solver( { "0x" + our_offer_store.store_id.hex(): { "new_root": "0x" + our_store_proofs[our_offer_store.store_id].proofs[0].root().hex(), "dependencies": [ { "launcher_id": "0x" + their_offer_store.store_id.hex(), "values_to_prove": [ "0x" + leaf_hash(key=entry.key, value=entry.value).hex() for entry in their_offer_store.inclusions ], } for their_offer_store in taker ], } for our_offer_store in maker } ) res = await self.wallet_rpc.create_offer_for_ids( CreateOfferForIDs( offer=offer_dict, solver=solver, driver_dict={}, fee=fee, validate_only=False, ), # TODO: probably shouldn't be default but due to peculiarities in the RPC, we're using a stop gap. # This is not a change in behavior, the default was already implicit. tx_config=DEFAULT_TX_CONFIG, ) offer = Offer( trade_id=res.trade_record.trade_id, offer=bytes(res.offer), taker=taker, maker=tuple(our_store_proofs.values()), ) # being extra careful and verifying the offer before returning it trading_offer = TradingOffer.from_bytes(offer.offer) summary = await DataLayerWallet.get_offer_summary(offer=trading_offer) verify_offer(maker=offer.maker, taker=offer.taker, summary=summary) return offer async def take_offer( self, offer_bytes: bytes, taker: tuple[OfferStore, ...], maker: tuple[StoreProofs, ...], fee: uint64, ) -> TradeRecord: async with self.data_store.transaction(): our_store_proofs = await self.process_offered_stores(offer_stores=taker) offer = TradingOffer.from_bytes(offer_bytes) summary = await DataLayerWallet.get_offer_summary(offer=offer) verify_offer(maker=maker, taker=taker, summary=summary) all_store_proofs: dict[bytes32, StoreProofs] = { store_proofs.proofs[0].root(): store_proofs for store_proofs in [*maker, *our_store_proofs.values()] } proofs_of_inclusion: list[tuple[str, str, list[str]]] = [] for root, store_proofs in all_store_proofs.items(): for proof in store_proofs.proofs: layers = [ ProofOfInclusionLayer( combined_hash=layer.combined_hash, other_hash_side=layer.other_hash_side, other_hash=layer.other_hash, ) for layer in proof.layers ] proof_of_inclusion = ProofOfInclusion(node_hash=proof.node_hash, layers=layers) sibling_sides_integer = calculate_sibling_sides_integer(proof_of_inclusion) proofs_of_inclusion.append( ( root.hex(), str(sibling_sides_integer), ["0x" + sibling_hash.hex() for sibling_hash in collect_sibling_hashes(proof_of_inclusion)], ) ) solver = Solver( { "proofs_of_inclusion": proofs_of_inclusion, **{ "0x" + our_offer_store.store_id.hex(): { "new_root": "0x" + root.hex(), "dependencies": [ { "launcher_id": "0x" + their_offer_store.store_id.hex(), "values_to_prove": [ "0x" + entry.node_hash.hex() for entry in their_offer_store.proofs ], } for their_offer_store in maker ], } for our_offer_store in taker }, } ) # 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. trade_record = ( await self.wallet_rpc.take_offer( TakeOffer(offer=offer.to_bech32(), solver=solver, fee=fee, push=True), # TODO: probably shouldn't be default but due to peculiarities in the RPC, we're using a stop gap. # This is not a change in behavior, the default was already implicit. tx_config=DEFAULT_TX_CONFIG, ) ).trade_record return trade_record async def cancel_offer(self, trade_id: bytes32, secure: bool, fee: uint64) -> None: store_ids: list[bytes32] = [] if not secure: trade_record = ( await self.wallet_rpc.get_offer(GetOffer(trade_id=trade_id, file_contents=True)) ).trade_record trading_offer = TradingOffer.from_bytes(trade_record.offer) summary = await DataLayerWallet.get_offer_summary(offer=trading_offer) store_ids = [offered.launcher_id for offered in summary.offered] await self.wallet_rpc.cancel_offer( CancelOffer(trade_id=trade_id, secure=secure, fee=fee, push=True), # TODO: probably shouldn't be default but due to peculiarities in the RPC, we're using a stop gap. # This is not a change in behavior, the default was already implicit. tx_config=DEFAULT_TX_CONFIG, ) if not secure: for store_id in store_ids: await self.data_store.clear_pending_roots(store_id=store_id) async def get_sync_status(self, store_id: bytes32) -> SyncStatus: await self._update_confirmation_status(store_id=store_id) if not await self.data_store.store_id_exists(store_id=store_id): raise Exception(f"No store id stored in the local database for {store_id}") root = await self.data_store.get_tree_root(store_id=store_id) singleton_record = ( await self.wallet_rpc.dl_latest_singleton(DLLatestSingleton(launcher_id=store_id, only_confirmed=True)) ).singleton if singleton_record is None: raise Exception(f"No singleton found for {store_id}") return SyncStatus( root_hash=self.none_bytes if root.node_hash is None else root.node_hash, generation=root.generation, target_root_hash=singleton_record.root, target_generation=singleton_record.generation, ) async def get_uploaders(self, store_id: bytes32) -> list[PluginRemote]: uploaders = [] for uploader in self.uploaders: async with aiohttp.ClientSession() as session: try: async with session.post( uploader.url + "/handle_upload", json={"store_id": store_id.hex()}, headers=uploader.headers, ) as response: res_json = await response.json() if res_json["handle_upload"]: uploaders.append(uploader) except Exception as e: self.log.error(f"get_uploader could not get response {e}") return uploaders async def check_plugins(self) -> PluginStatus: coros = [get_plugin_info(plugin_remote=plugin) for plugin in {*self.uploaders, *self.downloaders}] results = dict(await asyncio.gather(*coros)) unknown = { "name": "unknown", "version": "unknown", "instance": "unknown", } uploader_status = {uploader.url: results.get(uploader, unknown) for uploader in self.uploaders} downloader_status = {downloader.url: results.get(downloader, unknown) for downloader in self.downloaders} return PluginStatus(uploaders=uploader_status, downloaders=downloader_status)