Farmer solver networking (#20019)

* farmer connects to solver

* default solver peer

* configure solver peer

* solver trusted peers

* minor fixes

* connect to solver cli

* fix and rename fixture

* pr fixes

* load new service config if missing

* fix failling test

* fix test

Author: almogdepaz

chia/_tests/conftest.py

@@ -92,7 +92,7 @@
 from chia_rs.sized_ints import uint128
 
 from chia._tests.environments.wallet import WalletEnvironment, WalletState, WalletTestFramework
-from chia._tests.util.setup_nodes import setup_farmer_multi_harvester
+from chia._tests.util.setup_nodes import setup_farmer_solver_multi_harvester
 from chia.full_node.full_node_rpc_client import FullNodeRpcClient
 from chia.simulator.block_tools import BlockTools, create_block_tools_async, test_constants
 from chia.simulator.keyring import TempKeyring
@@ -866,7 +866,7 @@ async def farmer_one_harvester_simulator_wallet(
     ]
 ]:
     async with setup_simulators_and_wallets_service(1, 1, blockchain_constants) as (nodes, wallets, bt):
-        async with setup_farmer_multi_harvester(bt, 1, tmp_path, bt.constants, start_services=True) as (
+        async with setup_farmer_solver_multi_harvester(bt, 1, tmp_path, bt.constants, start_services=True) as (
             harvester_services,
             farmer_service,
             _,
@@ -879,7 +879,9 @@ async def farmer_one_harvester_simulator_wallet(
 
 @pytest.fixture(scope="function")
 async def farmer_one_harvester(tmp_path: Path, get_b_tools: BlockTools) -> AsyncIterator[FarmerOneHarvester]:
-    async with setup_farmer_multi_harvester(get_b_tools, 1, tmp_path, get_b_tools.constants, start_services=True) as _:
+    async with setup_farmer_solver_multi_harvester(
+        get_b_tools, 1, tmp_path, get_b_tools.constants, start_services=True
+    ) as _:
         yield _
 
 
@@ -890,37 +892,41 @@ async def farmer_one_harvester(tmp_path: Path, get_b_tools: BlockTools) -> Async
 async def farmer_one_harvester_solver(
     tmp_path: Path, get_b_tools: BlockTools
 ) -> AsyncIterator[FarmerOneHarvesterSolver]:
-    async with setup_farmer_multi_harvester(get_b_tools, 1, tmp_path, get_b_tools.constants, start_services=True) as (
-        harvester_services,
-        farmer_service,
-        bt,
-    ):
-        farmer_peer = UnresolvedPeerInfo(bt.config["self_hostname"], farmer_service._server.get_port())
-        async with setup_solver(tmp_path / "solver", bt, bt.constants, farmer_peer=farmer_peer) as solver_service:
+    async with setup_solver(tmp_path / "solver", get_b_tools, get_b_tools.constants) as solver_service:
+        solver_peer = UnresolvedPeerInfo(get_b_tools.config["self_hostname"], solver_service._server.get_port())
+        async with setup_farmer_solver_multi_harvester(
+            get_b_tools, 1, tmp_path, get_b_tools.constants, start_services=True, solver_peer=solver_peer
+        ) as (harvester_services, farmer_service, bt):
             yield harvester_services, farmer_service, solver_service, bt
 
 
 @pytest.fixture(scope="function")
 async def farmer_one_harvester_not_started(
     tmp_path: Path, get_b_tools: BlockTools
 ) -> AsyncIterator[tuple[list[HarvesterService], FarmerService, BlockTools]]:
-    async with setup_farmer_multi_harvester(get_b_tools, 1, tmp_path, get_b_tools.constants, start_services=False) as _:
+    async with setup_farmer_solver_multi_harvester(
+        get_b_tools, 1, tmp_path, get_b_tools.constants, start_services=False
+    ) as _:
         yield _
 
 
 @pytest.fixture(scope="function")
 async def farmer_two_harvester_not_started(
     tmp_path: Path, get_b_tools: BlockTools
 ) -> AsyncIterator[tuple[list[HarvesterService], FarmerService, BlockTools]]:
-    async with setup_farmer_multi_harvester(get_b_tools, 2, tmp_path, get_b_tools.constants, start_services=False) as _:
+    async with setup_farmer_solver_multi_harvester(
+        get_b_tools, 2, tmp_path, get_b_tools.constants, start_services=False
+    ) as _:
         yield _
 
 
 @pytest.fixture(scope="function")
 async def farmer_three_harvester_not_started(
     tmp_path: Path, get_b_tools: BlockTools
 ) -> AsyncIterator[tuple[list[HarvesterService], FarmerService, BlockTools]]:
-    async with setup_farmer_multi_harvester(get_b_tools, 3, tmp_path, get_b_tools.constants, start_services=False) as _:
+    async with setup_farmer_solver_multi_harvester(
+        get_b_tools, 3, tmp_path, get_b_tools.constants, start_services=False
+    ) as _:
         yield _
 
 
@@ -1301,10 +1307,13 @@ async def farmer_harvester_2_simulators_zero_bits_plot_filter(
             )
             for index in range(len(bts))
         ]
-
-        [harvester_service], farmer_service, _ = await async_exit_stack.enter_async_context(
-            setup_farmer_multi_harvester(bt, 1, tmp_path, bt.constants, start_services=True)
-        )
+        async with setup_solver(tmp_path / "solver", bt, bt.constants) as solver_service:
+            solver_peer = UnresolvedPeerInfo(bt.config["self_hostname"], solver_service._server.get_port())
+            [harvester_service], farmer_service, _ = await async_exit_stack.enter_async_context(
+                setup_farmer_solver_multi_harvester(
+                    bt, 1, tmp_path, bt.constants, start_services=True, solver_peer=solver_peer
+                )
+            )
 
         yield farmer_service, harvester_service, simulators[0], simulators[1], bt
 

chia/_tests/core/test_farmer_harvester_rpc.py

@@ -30,11 +30,15 @@
     plot_matches_filter,
 )
 from chia.farmer.farmer_rpc_client import FarmerRpcClient
+from chia.farmer.farmer_service import FarmerService
+from chia.harvester.harvester_service import HarvesterService
 from chia.plot_sync.receiver import Receiver, get_list_or_len
 from chia.plotting.util import add_plot_directory
 from chia.protocols import farmer_protocol
 from chia.protocols.harvester_protocol import Plot
-from chia.simulator.block_tools import get_plot_dir
+from chia.rpc.rpc_client import ResponseFailureError
+from chia.simulator.block_tools import BlockTools, get_plot_dir
+from chia.solver.solver_service import SolverService
 from chia.util.bech32m import decode_puzzle_hash, encode_puzzle_hash
 from chia.util.config import load_config, lock_and_load_config, save_config
 from chia.util.hash import std_hash
@@ -503,3 +507,32 @@ async def assert_added(path: Path) -> None:
     added_directories = await harvester_rpc_client.get_plot_directories()
     assert str(test_path) in added_directories
     assert str(test_path_other) in added_directories
+
+
+@pytest.mark.anyio
+async def test_farmer_connect_to_solver(
+    farmer_one_harvester_solver: tuple[list[HarvesterService], FarmerService, SolverService, BlockTools],
+) -> None:
+    _, farmer_service, solver_service, bt = farmer_one_harvester_solver
+    assert farmer_service.rpc_server is not None
+    farmer_rpc_client = await FarmerRpcClient.create(
+        bt.config["self_hostname"],
+        farmer_service.rpc_server.listen_port,
+        farmer_service.root_path,
+        farmer_service.config,
+    )
+
+    try:
+        # Test successful connection to existing solver
+        solver_host = bt.config["self_hostname"]
+        solver_port = solver_service._server.get_port()
+        result = await farmer_rpc_client.connect_to_solver(solver_host, solver_port)
+        assert result["success"] is True
+
+        # Test connection failure to non-existent solver
+        with pytest.raises(ResponseFailureError) as exc_info:
+            await farmer_rpc_client.connect_to_solver("localhost", 65000)
+        assert "Could not connect to solver" in exc_info.value.response["error"]
+    finally:
+        farmer_rpc_client.close()
+        await farmer_rpc_client.await_closed()

chia/_tests/farmer_harvester/test_filter_prefix_bits.py

@@ -11,7 +11,7 @@
 
 from chia._tests.conftest import ConsensusMode
 from chia._tests.core.test_farmer_harvester_rpc import wait_for_plot_sync
-from chia._tests.util.setup_nodes import setup_farmer_multi_harvester
+from chia._tests.util.setup_nodes import setup_farmer_solver_multi_harvester
 from chia._tests.util.time_out_assert import time_out_assert
 from chia.farmer.farmer_api import FarmerAPI
 from chia.farmer.farmer_rpc_client import FarmerRpcClient
@@ -60,7 +60,7 @@ async def have_connections() -> bool:
     )
     new_config = local_b_tools._config
     local_b_tools.change_config(new_config)
-    async with setup_farmer_multi_harvester(
+    async with setup_farmer_solver_multi_harvester(
         local_b_tools, 1, tmp_path, local_b_tools.constants, start_services=True
     ) as (harvesters, farmer_service, _):
         harvester_service = harvesters[0]

chia/_tests/farmer_harvester/test_third_party_harvesters.py

@@ -84,17 +84,16 @@ async def test_harvester_receive_source_signing_data(
     full_node_1: FullNode = full_node_service_1._node
     full_node_2: FullNode = full_node_service_2._node
 
+    await time_out_assert(60, node_type_connected, True, farmer.server, NodeType.HARVESTER)
     # Connect peers to each other
     farmer_service.add_peer(
         UnresolvedPeerInfo(str(full_node_service_2.self_hostname), full_node_service_2._server.get_port())
     )
+    await time_out_assert(60, node_type_connected, True, farmer.server, NodeType.FULL_NODE)
     full_node_service_2.add_peer(
         UnresolvedPeerInfo(str(full_node_service_1.self_hostname), full_node_service_1._server.get_port())
     )
-
-    await wait_until_node_type_connected(farmer.server, NodeType.FULL_NODE)
-    await wait_until_node_type_connected(farmer.server, NodeType.HARVESTER)  # Should already be connected
-    await wait_until_node_type_connected(full_node_1.server, NodeType.FULL_NODE)
+    await time_out_assert(60, node_type_connected, True, full_node_1.server, NodeType.FULL_NODE)
 
     # Prepare test data
     blocks: list[FullBlock]
@@ -429,12 +428,11 @@ def log_has_new_text() -> bool:
     return False
 
 
-async def wait_until_node_type_connected(server: ChiaServer, node_type: NodeType) -> WSChiaConnection:
-    while True:
-        for peer in server.all_connections.values():
-            if peer.connection_type == node_type.value:
-                return peer
-        await asyncio.sleep(1)
+def node_type_connected(server: ChiaServer, node_type: NodeType) -> bool:
+    for peer in server.all_connections.values():
+        if peer.connection_type == node_type.value:
+            return True
+    return False
 
 
 def decode_sp(

chia/_tests/util/setup_nodes.py

@@ -303,13 +303,14 @@ async def setup_simulators_and_wallets_inner(
 
 
 @asynccontextmanager
-async def setup_farmer_multi_harvester(
+async def setup_farmer_solver_multi_harvester(
     block_tools: BlockTools,
     harvester_count: int,
     temp_dir: Path,
     consensus_constants: ConsensusConstants,
     *,
     start_services: bool,
+    solver_peer: Optional[UnresolvedPeerInfo] = None,
 ) -> AsyncIterator[tuple[list[HarvesterService], FarmerService, BlockTools]]:
     async with AsyncExitStack() as async_exit_stack:
         farmer_service = await async_exit_stack.enter_async_context(
@@ -320,6 +321,7 @@ async def setup_farmer_multi_harvester(
                 consensus_constants,
                 port=uint16(0),
                 start_service=start_services,
+                solver_peer=solver_peer,
             )
         )
         if start_services:
@@ -342,74 +344,6 @@ async def setup_farmer_multi_harvester(
         yield harvester_services, farmer_service, block_tools
 
 
-@asynccontextmanager
-async def setup_farmer_multi_harvester_with_solver(
-    block_tools: BlockTools,
-    harvester_count: int,
-    temp_dir: Path,
-    consensus_constants: ConsensusConstants,
-    *,
-    start_services: bool,
-) -> AsyncIterator[tuple[list[HarvesterService], FarmerService, SolverService, BlockTools]]:
-    async with AsyncExitStack() as async_exit_stack:
-        farmer_service = await async_exit_stack.enter_async_context(
-            setup_farmer(
-                block_tools,
-                temp_dir / "farmer",
-                block_tools.config["self_hostname"],
-                consensus_constants,
-                port=uint16(0),
-                start_service=start_services,
-            )
-        )
-        if start_services:
-            farmer_peer = UnresolvedPeerInfo(block_tools.config["self_hostname"], farmer_service._server.get_port())
-        else:
-            farmer_peer = None
-        harvester_services = [
-            await async_exit_stack.enter_async_context(
-                setup_harvester(
-                    block_tools,
-                    temp_dir / f"harvester_{i}",
-                    farmer_peer,
-                    consensus_constants,
-                    start_service=start_services,
-                )
-            )
-            for i in range(harvester_count)
-        ]
-
-        # Setup solver with farmer peer - CRITICAL: use same BlockTools root path for SSL CA consistency
-        solver_service = await async_exit_stack.enter_async_context(
-            setup_solver(
-                temp_dir / "solver",  # Use temp_dir like harvester, not block_tools.root_path
-                block_tools,  # Pass BlockTools so SSL CA can be consistent
-                consensus_constants,
-                start_service=start_services,
-                farmer_peer=farmer_peer,
-            )
-        )
-
-        # Wait for farmer to be fully started before expecting solver connection
-        if start_services:
-            import asyncio
-
-            # Wait for farmer to be fully initialized
-            timeout = 30
-            for i in range(timeout):
-                if farmer_service._node.started:
-                    print(f"Farmer fully started after {i} seconds")
-                    break
-                await asyncio.sleep(1)
-            else:
-                print(f"WARNING: Farmer not started after {timeout} seconds")
-
-            # Give solver additional time to connect
-            await asyncio.sleep(3)
-
-        yield harvester_services, farmer_service, solver_service, block_tools
-
-
 @asynccontextmanager
 async def setup_full_system(
     consensus_constants: ConsensusConstants,

chia/cmds/configure.py

@@ -22,6 +22,7 @@
 def configure(
     root_path: Path,
     set_farmer_peer: str,
+    set_solver_peer: str,
     set_node_introducer: str,
     set_fullnode_port: str,
     set_harvester_port: str,
@@ -35,6 +36,7 @@ def configure(
     crawler_minimum_version_count: Optional[int],
     seeder_domain_name: str,
     seeder_nameserver: str,
+    set_solver_trusted_peers_only: str,
 ) -> None:
     config_yaml = "config.yaml"
     with lock_and_load_config(root_path, config_yaml, fill_missing_services=True) as config:
@@ -59,6 +61,25 @@ def configure(
                 change_made = True
             except ValueError:
                 print("Farmer address must be in format [IP:Port]")
+        if set_solver_peer:
+            try:
+                host, port = parse_host_port(set_solver_peer)
+                # ensure farmer section has solver_peers field
+                if "solver_peers" not in config["farmer"]:
+                    config["farmer"]["solver_peers"] = []
+                # Set single solver peer (overrides any existing)
+                config["farmer"]["solver_peers"] = [{"host": host, "port": port}]
+                print(f"Solver peer updated to {host}:{port}")
+                change_made = True
+            except ValueError:
+                print("Solver address must be in format [IP:Port]")
+        if set_solver_trusted_peers_only:
+            config["solver"]["trusted_peers_only"] = str2bool(set_solver_trusted_peers_only)
+            if str2bool(set_solver_trusted_peers_only):
+                print("Solver will only accept trusted peer connections")
+            else:
+                print("Solver will accept connections from all peers")
+            change_made = True
         if set_fullnode_port:
             config["full_node"]["port"] = int(set_fullnode_port)
             config["full_node"]["introducer_peer"]["port"] = int(set_fullnode_port)
@@ -237,6 +258,12 @@ def configure(
 )
 @click.option("--set-node-introducer", help="Set the introducer for node - IP:Port", type=str)
 @click.option("--set-farmer-peer", help="Set the farmer peer for harvester - IP:Port", type=str)
+@click.option("--set-solver-peer", help="Set the solver peer for farmer - IP:Port", type=str)
+@click.option(
+    "--set-solver-trusted-peers-only",
+    help="Enable/disable trusted peer requirement for solver connections",
+    type=click.Choice(["true", "t", "false", "f"]),
+)
 @click.option(
     "--set-fullnode-port",
     help="Set the port to use for the fullnode, useful for testing",
@@ -292,6 +319,7 @@ def configure(
 def configure_cmd(
     ctx: click.Context,
     set_farmer_peer: str,
+    set_solver_peer: str,
     set_node_introducer: str,
     set_fullnode_port: str,
     set_harvester_port: str,
@@ -305,10 +333,12 @@ def configure_cmd(
     crawler_minimum_version_count: int,
     seeder_domain_name: str,
     seeder_nameserver: str,
+    set_solver_trusted_peers_only: str,
 ) -> None:
     configure(
         ChiaCliContext.set_default(ctx).root_path,
         set_farmer_peer,
+        set_solver_peer,
         set_node_introducer,
         set_fullnode_port,
         set_harvester_port,
@@ -322,4 +352,5 @@ def configure_cmd(
         crawler_minimum_version_count,
         seeder_domain_name,
         seeder_nameserver,
+        set_solver_trusted_peers_only,
     )