Dl plugin service (#14883)

* generic downloader

* add s3 Downloader

* tests

* pre commit

* create uploader protocol, factor out upload code

* fix pre commit

* add s3 uploader

* add add helper for testing with real bucket

* typing

* lint

* logs

* start service separation

* working aiohttp s3 service

* lint

* multiple instances config

* break if failed to write

* better error handling

* update from conf before checking

* lint

* redundant config

* also update bucket

* improve exception handling

* remove old tests

* uae multiple uploaders

* Update chia/data_layer/data_layer.py

pr comments

Co-authored-by: Kyle Altendorf <sda@fstab.net>

* pr comments

* Apply typos fixes from code review

Co-authored-by: Kyle Altendorf <sda@fstab.net>

* Update chia/data_layer/s3_plugin_service.py

Co-authored-by: Kyle Altendorf <sda@fstab.net>

* scheme in conf

* use byte32 for store_ids list

* use dataclass instead of tuple for return value

* initial config handling for uploaders and downloaders

---------

Co-authored-by: Kyle Altendorf <sda@fstab.net>
Co-authored-by: Earle Lowe <30607889+emlowe@users.noreply.github.com>
Co-authored-by: Earle Lowe <e.lowe@chia.net>

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

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

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" ]

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()

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()
 

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
 

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