diff --git a/chia/data_layer/data_layer.py b/chia/data_layer/data_layer.py index e995e6b4f7..7c03ea92cc 100644 --- a/chia/data_layer/data_layer.py +++ b/chia/data_layer/data_layer.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import logging +import os import random import time import traceback @@ -58,6 +59,8 @@ class DataLayer: none_bytes: bytes32 lock: asyncio.Lock _server: Optional[ChiaServer] + downloaders: List[str] + uploaders: List[str] @property def server(self) -> ChiaServer: @@ -73,6 +76,8 @@ class DataLayer: config: Dict[str, Any], root_path: Path, wallet_rpc_init: Awaitable[WalletRpcClient], + downloaders: List[str], + uploaders: List[str], # dont add FilesystemUploader to this, it is the default uploader name: Optional[str] = None, ): if name == "": @@ -96,6 +101,8 @@ class DataLayer: self.none_bytes = bytes32([0] * 32) self.lock = asyncio.Lock() self._server = None + self.downloaders = downloaders + self.uploaders = uploaders def _set_state_changed_callback(self, callback: StateChangedProtocol) -> None: self.state_changed_callback = callback @@ -352,6 +359,7 @@ class DataLayer: random.shuffle(servers_info) for server_info in servers_info: url = server_info.url + root = await self.data_store.get_tree_root(tree_id=tree_id) if root.generation > singleton_record.generation: self.log.info( @@ -375,7 +383,6 @@ class DataLayer: min_generation=uint32(root.generation + 1), max_generation=singleton_record.generation, ) - try: timeout = self.config.get("client_timeout", 15) proxy_url = self.config.get("proxy_url", None) @@ -389,6 +396,7 @@ class DataLayer: timeout, self.log, proxy_url, + await self.get_downloader(url), ) if success: self.log.info( @@ -404,7 +412,21 @@ class DataLayer: except Exception as e: self.log.warning(f"Exception while downloading files for {tree_id}: {e} {traceback.format_exc()}.") + async def get_downloader(self, url: str) -> Optional[str]: + request_json = {"url": url} + for d in self.downloaders: + async with aiohttp.ClientSession() as session: + try: + async with session.post(d + "/check_url", json=request_json) as response: + res_json = await response.json() + if res_json["handles_url"]: + return d + except Exception as e: + self.log.error(f"get_downloader could not get response: {type(e).__name__}: {e}") + return None + async def upload_files(self, tree_id: bytes32) -> None: + uploaders = await self.get_uploaders(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"Upload files: no on-chain record for {tree_id}.") @@ -417,12 +439,28 @@ class DataLayer: # 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(tree_id=tree_id, generation=publish_generation) - while publish_generation > 0 and await write_files_for_root( - self.data_store, - tree_id, - root, - self.server_files_location, - ): + while publish_generation > 0: + write_file_result = await write_files_for_root(self.data_store, tree_id, root, self.server_files_location) + if not write_file_result.result: + self.log.error("failed to write files") + break + try: + if uploaders is not None and len(uploaders) > 0: + request_json = { + "id": tree_id.hex(), + "full_tree_path": str(write_file_result.full_tree), + "diff_path": str(write_file_result.diff_tree), + } + for uploader in uploaders: + async with aiohttp.ClientSession() as session: + async with session.post(uploader + "/upload", json=request_json) as response: + res_json = await response.json() + if not res_json["uploaded"]: + break # todo this will retry all uploaders + except Exception as e: + self.log.debug(f"failed to upload files, clean local disc {e}") + 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(tree_id=tree_id, generation=publish_generation) @@ -802,3 +840,18 @@ class DataLayer: target_root_hash=singleton_record.root, target_generation=singleton_record.generation, ) + + async def get_uploaders(self, tree_id: bytes32) -> List[str]: + uploaders = [] + for uploader in self.uploaders: + async with aiohttp.ClientSession() as session: + try: + async with session.post( + "http://" + uploader + "/check_store_id", json={"id": tree_id.hex()} + ) as response: + res_json = await response.json() + if res_json["handles_store"]: + uploaders.append(uploader) + except Exception as e: + self.log.error(f"get_uploader could not get response {e}") + return uploaders diff --git a/chia/data_layer/download_data.py b/chia/data_layer/download_data.py index a1702d8a08..9f8eed7034 100644 --- a/chia/data_layer/download_data.py +++ b/chia/data_layer/download_data.py @@ -4,6 +4,7 @@ import asyncio import logging import os import time +from dataclasses import dataclass from pathlib import Path from typing import List, Optional @@ -86,13 +87,20 @@ async def insert_into_data_store_from_file( await data_store.insert_root_with_ancestor_table(tree_id=tree_id, node_hash=root_hash, status=Status.COMMITTED) +@dataclass +class WriteFilesResult: + result: bool + full_tree: Path + diff_tree: Path + + async def write_files_for_root( data_store: DataStore, tree_id: bytes32, root: Root, foldername: Path, overwrite: bool = False, -) -> bool: +) -> WriteFilesResult: if root.node_hash is not None: node_hash = root.node_hash else: @@ -124,7 +132,7 @@ async def write_files_for_root( except FileExistsError: pass - return written + return WriteFilesResult(written, filename_full_tree, filename_diff_tree) async def insert_from_delta_file( @@ -137,37 +145,23 @@ async def insert_from_delta_file( timeout: int, log: logging.Logger, proxy_url: str, + downloader: Optional[str], ) -> bool: for root_hash in root_hashes: timestamp = int(time.time()) existing_generation += 1 filename = get_delta_filename(tree_id, root_hash, existing_generation) - - try: + request_json = {"url": server_info.url, "client_folder": str(client_foldername), "filename": filename} + if downloader is None: + # use http downloader + if not await http_download(client_foldername, filename, proxy_url, server_info, timeout, log): + break + else: async with aiohttp.ClientSession() as session: - headers = {"accept-encoding": "gzip"} - async with session.get( - server_info.url + "/" + filename, headers=headers, timeout=timeout, proxy=proxy_url - ) as resp: - resp.raise_for_status() - size = int(resp.headers.get("content-length", 0)) - log.debug(f"Downloading delta file {filename}. Size {size} bytes.") - progress_byte = 0 - progress_percentage = "{:.0%}".format(0) - target_filename = client_foldername.joinpath(filename) - with target_filename.open(mode="wb") as f: - async for chunk, _ in resp.content.iter_chunks(): - f.write(chunk) - progress_byte += len(chunk) - new_percentage = "{:.0%}".format(progress_byte / size) - if new_percentage != progress_percentage: - progress_percentage = new_percentage - log.info(f"Downloading delta file {filename}. {progress_percentage} of {size} bytes.") - except Exception: - target_filename = client_foldername.joinpath(filename) - os.remove(target_filename) - await data_store.server_misses_file(tree_id, server_info, timestamp) - raise + async with session.post(downloader + "/download", json=request_json) as response: + res_json = await response.json() + if not res_json["downloaded"]: + break log.info(f"Successfully downloaded delta file {filename}.") try: @@ -200,3 +194,34 @@ async def insert_from_delta_file( raise return True + + +async def http_download( + client_folder: Path, + filename: str, + proxy_url: str, + server_info: ServerInfo, + timeout: int, + log: logging.Logger, +) -> bool: + async with aiohttp.ClientSession() as session: + headers = {"accept-encoding": "gzip"} + async with session.get( + server_info.url + "/" + filename, headers=headers, timeout=timeout, proxy=proxy_url + ) as resp: + resp.raise_for_status() + size = int(resp.headers.get("content-length", 0)) + log.debug(f"Downloading delta file {filename}. Size {size} bytes.") + progress_byte = 0 + progress_percentage = "{:.0%}".format(0) + target_filename = client_folder.joinpath(filename) + with target_filename.open(mode="wb") as f: + async for chunk, _ in resp.content.iter_chunks(): + f.write(chunk) + progress_byte += len(chunk) + new_percentage = "{:.0%}".format(progress_byte / size) + if new_percentage != progress_percentage: + progress_percentage = new_percentage + log.info(f"Downloading delta file {filename}. {progress_percentage} of {size} bytes.") + + return True diff --git a/chia/data_layer/s3_plugin_config.yml b/chia/data_layer/s3_plugin_config.yml new file mode 100644 index 0000000000..e5b86ef04c --- /dev/null +++ b/chia/data_layer/s3_plugin_config.yml @@ -0,0 +1,27 @@ +instance-1: + port: 8998 + aws_credentials: + access_key_id: "xxx" + secret_access_key: "xxx" + region: "xxx" + + store_ids: ["xxx"] + + urls: [ "xxx"] + + buckets: + chia-datalayer-test-bucket-2: ["xxx"] + +instance-2: + port: 8999 + aws_credentials: + access_key_id: "xxx" + secret_access_key: "xxx" + region: "xxx" + + store_ids: ["xxx", "xxx" ] + + urls: [ "xxx"] + + buckets: + chia-datalayer-test-bucket-1: [ "xxx", "xxx" ] diff --git a/chia/data_layer/s3_plugin_service.py b/chia/data_layer/s3_plugin_service.py new file mode 100644 index 0000000000..033d0cb78d --- /dev/null +++ b/chia/data_layer/s3_plugin_service.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import asyncio +import concurrent.futures +import functools +import logging +import sys +from pathlib import Path +from typing import Any, Dict, List +from urllib.parse import urlparse + +import boto3 as boto3 +import yaml +from aiohttp import web +from botocore.exceptions import ClientError + +from chia.types.blockchain_format.sized_bytes import bytes32 + +log = logging.getLogger(__name__) + + +class S3Plugin: + boto_client: boto3.client + port: int + region: str + aws_access_key_id: str + aws_secret_access_key: str + store_ids: List[bytes32] + bukets: Dict[str, List[str]] + urls: List[str] + instance_name: str + + def __init__( + self, + region: str, + aws_access_key_id: str, + aws_secret_access_key: str, + store_ids: List[bytes32], + buckets: Dict[str, List[str]], + urls: List[str], + instance_name: str, + ): + self.boto_client = boto3.client( + "s3", + region_name=region, + aws_access_key_id=aws_access_key_id, + aws_secret_access_key=aws_secret_access_key, + ) + self.store_ids = store_ids + self.buckets = buckets + self.urls = urls + self.instance_name = instance_name + + async def check_store_id(self, request: web.Request) -> web.Response: + self.update_instance_from_config() + try: + data = await request.json() + except Exception as e: + print(f"failed parsing request {request} {e}") + return web.json_response({"handles_url": False}) + store_id = bytes32.from_hexstr(data["id"]) + if store_id in self.store_ids: + return web.json_response({"handles_store": True}) + return web.json_response({"handles_store": False}) + + async def upload(self, request: web.Request) -> web.Response: + try: + data = await request.json() + store_id = bytes32.from_hexstr(data["id"]) + bucket = self.get_bucket(store_id) + full_tree_path = Path(data["full_tree_path"]) + diff_path = Path(data["diff_path"]) + try: + with concurrent.futures.ThreadPoolExecutor() as pool: + await asyncio.get_running_loop().run_in_executor( + pool, + functools.partial(self.boto_client.upload_file, full_tree_path, bucket, full_tree_path.name), + ) + await asyncio.get_running_loop().run_in_executor( + pool, functools.partial(self.boto_client.upload_file, diff_path, bucket, diff_path.name) + ) + except ClientError as e: + print(f"failed uploading file to aws {e}") + return web.json_response({"uploaded": False}) + except Exception as e: + print(f"failed handling request {request} {e}") + return web.json_response({"handles_url": False}) + return web.json_response({"uploaded": True}) + + async def check_url(self, request: web.Request) -> web.Response: + self.update_instance_from_config() + try: + data = await request.json() + except Exception as e: + print(f"failed parsing request {request} {e}") + return web.json_response({"handles_url": False}) + parse_result = urlparse(data["url"]) + if parse_result.scheme == "s3" and data["url"] in self.urls: + return web.json_response({"handles_url": True}) + return web.json_response({"handles_url": False}) + + async def download(self, request: web.Request) -> web.Response: + try: + data = await request.json() + url = data["url"] + client_folder = Path(data["client_folder"]) + filename = data["filename"] + parse_result = urlparse(url) + bucket = parse_result.netloc + target_filename = client_folder.joinpath(filename) + # Create folder for parent directory + target_filename.parent.mkdir(parents=True, exist_ok=True) + with concurrent.futures.ThreadPoolExecutor() as pool: + await asyncio.get_running_loop().run_in_executor( + pool, functools.partial(self.boto_client.download_file, bucket, filename, str(target_filename)) + ) + except Exception as e: + print(f"failed parsing request {request} {e}") + return web.json_response({"downloaded": False}) + return web.json_response({"downloaded": True}) + + def get_bucket(self, store_id: bytes32) -> str: + for bucket in self.buckets: + if store_id.hex() in self.buckets[bucket]: + return bucket + raise Exception(f"bucket not found for store id {store_id.hex()}") + + def update_instance_from_config(self) -> None: + config = load_config(self.instance_name) + store_ids = config["store_ids"] + buckets: Dict[str, List[str]] = config["buckets"] + urls = config["urls"] + self.buckets = buckets + self.store_ids = store_ids + self.urls = urls + + +def make_app(config: Dict[str, Any], instance_name: str): # type: ignore + region = config["aws_credentials"]["region"] + aws_access_key_id = config["aws_credentials"]["access_key_id"] + aws_secret_access_key = config["aws_credentials"]["secret_access_key"] + store_ids = [] + for store in config["store_ids"]: + store_ids.append(bytes32.from_hexstr(store)) + buckets: Dict[str, List[str]] = config["buckets"] + urls = config["urls"] + s3_client = S3Plugin(region, aws_access_key_id, aws_secret_access_key, store_ids, buckets, urls, instance_name) + app = web.Application() + app.add_routes([web.post("/check_store_id", s3_client.check_store_id)]) + app.add_routes([web.post("/upload", s3_client.upload)]) + app.add_routes([web.post("/check_url", s3_client.check_url)]) + app.add_routes([web.post("/download", s3_client.download)]) + return app + + +def load_config(instance: str) -> Any: + with open("s3_plugin_config.yml", "r") as f: + full_config = yaml.safe_load(f) + return full_config[instance] + + +def run_server() -> None: + instance_name = sys.argv[1] + print(f"run instance {instance_name}") + config = load_config(instance_name) + port = config["port"] + web.run_app(make_app(config, instance_name), port=port) + + +# run this +run_server() diff --git a/chia/server/start_data_layer.py b/chia/server/start_data_layer.py index 805b5f8d3f..55b0b7d9d2 100644 --- a/chia/server/start_data_layer.py +++ b/chia/server/start_data_layer.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging import pathlib import sys -from typing import Any, Dict, Optional, cast +from typing import Any, Dict, List, Optional, cast from chia.data_layer.data_layer import DataLayer from chia.data_layer.data_layer_api import DataLayerAPI @@ -29,9 +29,15 @@ log = logging.getLogger(__name__) def create_data_layer_service( root_path: pathlib.Path, config: Dict[str, Any], + downloaders: List[str], + uploaders: List[str], # dont add FilesystemUploader to this, it is the default uploader wallet_service: Optional[Service[WalletNode]] = None, connect_to_daemon: bool = True, ) -> Service[DataLayer]: + if uploaders is None: + uploaders = [] + if downloaders is None: + downloaders = [] service_config = config[SERVICE_NAME] self_hostname = config["self_hostname"] wallet_rpc_port = service_config["wallet_peer"]["port"] @@ -42,7 +48,14 @@ def create_data_layer_service( wallet_root_path = wallet_service.root_path wallet_config = wallet_service.config wallet_rpc_init = WalletRpcClient.create(self_hostname, uint16(wallet_rpc_port), wallet_root_path, wallet_config) - data_layer = DataLayer(config=service_config, root_path=root_path, wallet_rpc_init=wallet_rpc_init) + + data_layer = DataLayer( + config=service_config, + root_path=root_path, + wallet_rpc_init=wallet_rpc_init, + downloaders=downloaders, + uploaders=uploaders, + ) # dont add Fil) api = DataLayerAPI(data_layer) network_id = service_config["selected_network"] rpc_port = service_config.get("rpc_port") @@ -86,7 +99,9 @@ async def async_main() -> int: overwrite=False, ) - service = create_data_layer_service(DEFAULT_ROOT_PATH, config) + uploaders: List[str] = config["data_layer"].get("uploaders", []) + downloaders: List[str] = config["data_layer"].get("downloaders", []) + service = create_data_layer_service(DEFAULT_ROOT_PATH, config, downloaders, uploaders) await service.setup_process_global_state() await service.run() diff --git a/chia/util/initial-config.yaml b/chia/util/initial-config.yaml index bbbf0aa10b..2b4f4d2608 100644 --- a/chia/util/initial-config.yaml +++ b/chia/util/initial-config.yaml @@ -107,23 +107,21 @@ chia_ssl_ca: crt: "config/ssl/ca/chia_ca.crt" key: "config/ssl/ca/chia_ca.key" - daemon_ssl: private_crt: "config/ssl/daemon/private_daemon.crt" private_key: "config/ssl/daemon/private_daemon.key" - # Controls logging of all servers (harvester, farmer, etc..). Each one can be overridden. logging: &logging - log_stdout: False # If True, outputs to stdout instead of a file + log_stdout: False # If True, outputs to stdout instead of a file log_filename: "log/debug.log" - log_level: "WARNING" # Can be CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET + log_level: "WARNING" # Can be CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET log_maxfilesrotation: 7 # Max files in rotation. Default value 7 if the key is not set log_maxbytesrotation: 52428800 # Max bytes logged before rotating logs log_use_gzip: False # Use gzip to compress rotated logs - log_syslog: False # If True, outputs to SysLog host and port specified - log_syslog_host: "localhost" # Send logging messages to a remote or local Unix syslog - log_syslog_port: 514 # UDP port of the remote or local Unix syslog + log_syslog: False # If True, outputs to SysLog host and port specified + log_syslog_host: "localhost" # Send logging messages to a remote or local Unix syslog + log_syslog_port: 514 # UDP port of the remote or local Unix syslog seeder: # The fake full node used for crawling will run on this port. @@ -179,7 +177,6 @@ harvester: batch_size: 300 # How many plot files the harvester processes before it waits batch_sleep_milliseconds batch_sleep_milliseconds: 1 # Milliseconds the harvester sleeps between batch processing - # If True use parallel reads in chiapos parallel_read: True @@ -192,8 +189,8 @@ harvester: recursive_plot_scan: False # If True the harvester scans plots recursively in the provided directories. ssl: - private_crt: "config/ssl/harvester/private_harvester.crt" - private_key: "config/ssl/harvester/private_harvester.key" + private_crt: "config/ssl/harvester/private_harvester.crt" + private_key: "config/ssl/harvester/private_harvester.key" private_ssl_ca: crt: "config/ssl/ca/private_ca.crt" @@ -210,7 +207,6 @@ pool: network_overrides: *network_overrides selected_network: *selected_network - farmer: # The farmer server (if run) will run on this port port: 8447 @@ -238,10 +234,10 @@ farmer: selected_network: *selected_network ssl: - private_crt: "config/ssl/farmer/private_farmer.crt" - private_key: "config/ssl/farmer/private_farmer.key" - public_crt: "config/ssl/farmer/public_farmer.crt" - public_key: "config/ssl/farmer/public_farmer.key" + private_crt: "config/ssl/farmer/private_farmer.crt" + private_key: "config/ssl/farmer/private_farmer.key" + public_crt: "config/ssl/farmer/public_farmer.crt" + public_key: "config/ssl/farmer/public_farmer.key" # Don't run this unless you want to run VDF clients on the local machine. timelord_launcher: @@ -252,7 +248,6 @@ timelord_launcher: process_count: 3 logging: *logging - timelord: # The timelord server (if run) will run on this port port: 8446 @@ -266,8 +261,8 @@ timelord: ips_estimate: - 150000 full_node_peer: - host: *self_hostname - port: 8444 + host: *self_hostname + port: 8444 # Maximum number of seconds allowed for a client to reconnect to the server. max_connection_time: 60 # The ip and port where the TCP clients will connect. @@ -306,10 +301,10 @@ timelord: rpc_port: 8557 ssl: - private_crt: "config/ssl/timelord/private_timelord.crt" - private_key: "config/ssl/timelord/private_timelord.key" - public_crt: "config/ssl/timelord/public_timelord.crt" - public_key: "config/ssl/timelord/public_timelord.key" + private_crt: "config/ssl/timelord/private_timelord.crt" + private_key: "config/ssl/timelord/private_timelord.key" + public_crt: "config/ssl/timelord/public_timelord.crt" + public_key: "config/ssl/timelord/public_timelord.key" full_node: # The full node server (if run) will run on this port @@ -425,14 +420,14 @@ full_node: - "chia.hoffmang.com" - "seeder.xchpool.org" farmer_peer: - host: *self_hostname - port: 8447 + host: *self_hostname + port: 8447 timelord_peer: - host: *self_hostname - port: 8446 + host: *self_hostname + port: 8446 introducer_peer: - host: introducer.chia.net # Chia AWS introducer IPv4/IPv6 - port: 8444 + host: introducer.chia.net # Chia AWS introducer IPv4/IPv6 + port: 8444 wallet_peer: host: *self_hostname port: 8449 @@ -445,14 +440,14 @@ full_node: 0ThisisanexampleNodeID7ff9d60f1c3fa270c213c0ad0cb89c01274634a7c3cb7: Does_not_matter ssl: - private_crt: "config/ssl/full_node/private_full_node.crt" - private_key: "config/ssl/full_node/private_full_node.key" - public_crt: "config/ssl/full_node/public_full_node.crt" - public_key: "config/ssl/full_node/public_full_node.key" + private_crt: "config/ssl/full_node/private_full_node.crt" + private_key: "config/ssl/full_node/private_full_node.key" + public_crt: "config/ssl/full_node/public_full_node.crt" + public_key: "config/ssl/full_node/public_full_node.key" use_chia_loop_policy: True ui: - # The ui node server (if run) will run on this port + # The ui node server (if run) will run on this port port: 8222 # Which port to use to communicate with the full node @@ -484,8 +479,8 @@ introducer: selected_network: *selected_network ssl: - public_crt: "config/ssl/full_node/public_full_node.crt" - public_key: "config/ssl/full_node/public_full_node.key" + public_crt: "config/ssl/full_node/public_full_node.crt" + public_key: "config/ssl/full_node/public_full_node.key" wallet: port: 8449 @@ -547,10 +542,10 @@ wallet: port: 8444 ssl: - private_crt: "config/ssl/wallet/private_wallet.crt" - private_key: "config/ssl/wallet/private_wallet.key" - public_crt: "config/ssl/wallet/public_wallet.crt" - public_key: "config/ssl/wallet/public_wallet.key" + private_crt: "config/ssl/wallet/private_wallet.crt" + private_key: "config/ssl/wallet/private_wallet.key" + public_crt: "config/ssl/wallet/public_wallet.crt" + public_key: "config/ssl/wallet/public_wallet.key" # Node IDs of trusted full node peers, for performing a fast trusted wallet sync trusted_peers: @@ -619,13 +614,14 @@ data_layer: # TODO: which of these are really appropriate? ssl: - private_crt: "config/ssl/data_layer/private_data_layer.crt" - private_key: "config/ssl/data_layer/private_data_layer.key" - public_crt: "config/ssl/data_layer/public_data_layer.crt" - public_key: "config/ssl/data_layer/public_data_layer.key" + private_crt: "config/ssl/data_layer/private_data_layer.crt" + private_key: "config/ssl/data_layer/private_data_layer.key" + public_crt: "config/ssl/data_layer/public_data_layer.crt" + public_key: "config/ssl/data_layer/public_data_layer.key" + uploaders: [] + downloaders: [] simulator: - # Should the simulator farm a block whenever a transaction is in mempool auto_farm: True diff --git a/tests/core/data_layer/test_data_rpc.py b/tests/core/data_layer/test_data_rpc.py index e743ef8957..9ca5e09860 100644 --- a/tests/core/data_layer/test_data_rpc.py +++ b/tests/core/data_layer/test_data_rpc.py @@ -53,7 +53,9 @@ async def init_data_layer( config["data_layer"]["rpc_port"] = 0 config["data_layer"]["database_path"] = str(db_path.joinpath("db.sqlite")) save_config(bt.root_path, "config.yaml", config) - service = create_data_layer_service(root_path=bt.root_path, config=config, wallet_service=wallet_service) + service = create_data_layer_service( + root_path=bt.root_path, config=config, wallet_service=wallet_service, downloaders=[], uploaders=[] + ) await service.start() try: yield service._api.data_layer