wallet server federation, client portion

lbry/conf.py

@@ -273,6 +273,36 @@ class Strings(ListSetting):
                 f"'{self.name}' must be a string."
 
 
+class KnownHubsList:
+
+    def __init__(self, config: 'Config', file_name: str = 'known_hubs.yml'):
+        self.file_name = file_name
+        self.path = os.path.join(config.wallet_dir, self.file_name)
+        self.hubs = []
+        if self.exists:
+            self.load()
+
+    @property
+    def exists(self):
+        return self.path and os.path.exists(self.path)
+
+    def load(self):
+        with open(self.path, 'r') as known_hubs_file:
+            raw = known_hubs_file.read()
+            self.hubs = yaml.safe_load(raw) or {}
+
+    def save(self):
+        with open(self.path, 'w') as known_hubs_file:
+            known_hubs_file.write(yaml.safe_dump(self.hubs, default_flow_style=False))
+
+    def append(self, hub: str):
+        self.hubs.append(hub)
+        return hub
+
+    def __iter__(self):
+        return iter(self.hubs)
+
+
 class EnvironmentAccess:
     PREFIX = 'LBRY_'
 
@@ -655,6 +685,7 @@ class Config(CLIConfig):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.set_default_paths()
+        self.known_hubs = KnownHubsList(self)
 
     def set_default_paths(self):
         if 'darwin' in sys.platform.lower():

lbry/wallet/network.py

@@ -293,6 +293,8 @@ class Network:
                 log.debug("get spv server features %s:%i", *client.server)
                 features = await client.send_request('server.features', [])
                 self.client, self.server_features = client, features
+                log.debug("discover other hubs %s:%i", *client.server)
+                peers = await client.send_request('server.peers.get', [])
                 log.info("subscribe to headers %s:%i", *client.server)
                 self._update_remote_height((await self.subscribe_headers(),))
                 self._on_connected_controller.add(True)

lbry/wallet/server/env.py

@@ -269,3 +269,6 @@ class Env:
             return self.PD_SELF
         else:
             return self.PD_ON
+
+    def peer_hubs(self):
+        return [hub.strip() for hub in self.default('PEER_HUBS', '').split(',')]

lbry/wallet/server/session.py

@@ -878,6 +878,7 @@ class LBRYElectrumX(SessionBase):
             'server.payment_address': cls.payment_address,
             'server.donation_address': cls.donation_address,
             'server.features': cls.server_features_async,
+            'server.peers.get': cls.peers_get,
             'server.peers.subscribe': cls.peers_subscribe,
             'server.version': cls.server_version,
             'blockchain.transaction.get_height': cls.transaction_get_height,
@@ -1082,6 +1083,10 @@ class LBRYElectrumX(SessionBase):
         """Add a peer (but only if the peer resolves to the source)."""
         return await self.peer_mgr.on_add_peer(features, self.peer_address())
 
+    async def peers_get(self):
+        """Return the server peers as a list of (ip, host, details) tuples."""
+        return self.env.peer_hubs()
+
     async def peers_subscribe(self):
         """Return the server peers as a list of (ip, host, details) tuples."""
         return self.peer_mgr.on_peers_subscribe(self.is_tor())

tests/integration/blockchain/test_wallet_server_sessions.py

@@ -113,3 +113,32 @@ class TestESSync(CommandTestCase):
         self.assertTrue(await make_es_index(db.search_index))
         await db.search_index.start()
         await resync()
+
+
+class TestHubDiscovery(CommandTestCase):
+
+    async def test_hub_discovery(self):
+        final_node = SPVNode(self.conductor.spv_module, node_number=2)
+        await final_node.start(self.blockchain)
+        self.addCleanup(final_node.stop)
+        final_node_host = f"{final_node.hostname}:{final_node.port}"
+
+        relay_node = SPVNode(self.conductor.spv_module, node_number=3)
+        await relay_node.start(self.blockchain, extraconf={"PEER_HUBS": final_node_host})
+        relay_node_host = f"{relay_node.hostname}:{relay_node.port}"
+        self.addCleanup(relay_node.stop)
+
+        self.assertEqual(list(self.daemon.conf.known_hubs), [])
+        self.daemon.conf.known_hubs.append(relay_node_host)
+
+        self.assertEqual(
+            self.daemon.ledger.network.client.server_address_and_port,
+            ('127.0.0.1', 50002)
+        )
+
+        await self.daemon.jsonrpc_wallet_reconnect()
+
+        self.assertEqual(
+            self.daemon.ledger.network.client.server_address_and_port,
+            ('127.0.0.1', 50003)
+        )

tests/unit/test_conf.py

@@ -255,3 +255,12 @@ class ConfigurationTests(unittest.TestCase):
         args = parser.parse_args(['--string-choice', 'c'])
         c = TestConfig.create_from_arguments(args)
         self.assertEqual("c", c.string_choice)
+
+    def test_known_hubs_list(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            c1 = Config(config=os.path.join(temp_dir, 'settings.yml'), wallet_dir=temp_dir)
+            self.assertEqual(list(c1.known_hubs), [])
+            c1.known_hubs.append('new.hub.io')
+            c1.known_hubs.save()
+            c2 = Config(config=os.path.join(temp_dir, 'settings.yml'), wallet_dir=temp_dir)
+            self.assertEqual(list(c2.known_hubs), ['new.hub.io'])