From 992b5bd7d1ddf88b928352116c7fa6b93202ba06 Mon Sep 17 00:00:00 2001 From: Dominic Letz Date: Mon, 13 Jul 2026 15:32:18 +0200 Subject: [PATCH 1/2] Harden Anvil restarts and test isolation so the full suite passes reliably. Retry RPC on NodeProxy disconnect, bootstrap Anvil via HTTP mining, restore selective chain-list cache refresh, and make PeerReadyEts safe when the ETS table is torn down during test service restarts. --- lib/network/peer_ready_ets.ex | 30 ++++++++++++--- lib/remote_chain/chain_list.ex | 37 +++++++++++++++++-- lib/remote_chain/rpc.ex | 18 ++++++++- test/chains/epoch_block_test.exs | 4 +- test/network/common_test.exs | 8 ++++ test/network/device_notify_e2e_test.exs | 6 +-- test/network/rpc_ws_ticket_billing_test.exs | 6 +-- .../rpc_ws_ticket_request_e2e_test.exs | 3 +- test/test_helper.exs | 30 ++++++++++++++- 9 files changed, 119 insertions(+), 23 deletions(-) diff --git a/lib/network/peer_ready_ets.ex b/lib/network/peer_ready_ets.ex index f957b44..2e2bb9f 100644 --- a/lib/network/peer_ready_ets.ex +++ b/lib/network/peer_ready_ets.ex @@ -14,18 +14,38 @@ defmodule Network.PeerReadyEts do :ets.new(@table, [:named_table, :set, :public, read_concurrency: true]) end + def ensure do + case :ets.info(@table) do + :undefined -> + :ets.new(@table, [:named_table, :set, :public, read_concurrency: true]) + + _ -> + :ok + end + end + def insert(address, pid) when is_binary(address) and is_pid(pid) do + ensure() :ets.insert(@table, {address, pid}) end def delete(address) when is_binary(address) do - :ets.delete(@table, address) + case :ets.info(@table) do + :undefined -> :ok + _ -> :ets.delete(@table, address) + end end def read do - @table - |> :ets.tab2list() - |> Enum.filter(fn {_address, pid} -> Process.alive?(pid) end) - |> Map.new() + case :ets.info(@table) do + :undefined -> + %{} + + _ -> + @table + |> :ets.tab2list() + |> Enum.filter(fn {_address, pid} -> Process.alive?(pid) end) + |> Map.new() + end end end diff --git a/lib/remote_chain/chain_list.ex b/lib/remote_chain/chain_list.ex index 63b89b6..f09ff81 100644 --- a/lib/remote_chain/chain_list.ex +++ b/lib/remote_chain/chain_list.ex @@ -136,10 +136,40 @@ defmodule RemoteChain.ChainList do defp refresh_chains() do file_path() |> File.read!() + |> load_chains_from_json!() + |> put_chains() + end + + @doc false + def refresh_chains(chains) when is_list(chains), do: put_chains(chains, only_cached: true) + + defp load_chains_from_json!(json) when is_binary(json) do + json |> Jason.decode!() - |> Map.new(fn %{"chainId" => id} = chain -> {id, chain} end) - |> Enum.each(fn {id, chain} -> - Globals.put(cache_key(id), chain) + |> chains_to_map() + end + + defp chains_to_map(chains) do + Map.new(chains, fn %{"chainId" => id} = chain -> {id, chain} end) + end + + defp put_chains(chains, opts \\ []) + + defp put_chains(chains, opts) when is_list(chains) do + chains + |> chains_to_map() + |> put_chains(opts) + end + + defp put_chains(chains_by_id, opts) when is_map(chains_by_id) do + only_cached? = Keyword.get(opts, :only_cached, false) + + Enum.each(chains_by_id, fn {id, chain} -> + key = cache_key(id) + + if not only_cached? or Globals.get(key) != nil do + Globals.put(key, chain) + end end) end @@ -163,6 +193,7 @@ defmodule RemoteChain.ChainList do # Just ensure it's valid JSON {:ok, _chains} = Jason.decode(json) File.write!(Diode.data_dir("chains.json"), json) + Globals.pop(@loaded_key) refresh_chains() :updated end diff --git a/lib/remote_chain/rpc.ex b/lib/remote_chain/rpc.ex index c65ca9e..479ae43 100644 --- a/lib/remote_chain/rpc.ex +++ b/lib/remote_chain/rpc.ex @@ -57,9 +57,23 @@ defmodule RemoteChain.RPC do end def rpc(chain, method, params) do + rpc_with_retry(chain, method, params, 2) + end + + defp rpc_with_retry(chain, method, params, retries_left) do case RemoteChain.NodeProxy.rpc(chain, method, params) do - %{"result" => result} -> {:ok, result} - %{"error" => error} -> {:error, error} + %{"result" => result} -> + {:ok, result} + + %{"error" => error} -> + {:error, error} + + {:error, :disconnect} when retries_left > 0 -> + Process.sleep(200) + rpc_with_retry(chain, method, params, retries_left - 1) + + {:error, :disconnect} -> + {:error, :disconnect} end end diff --git a/test/chains/epoch_block_test.exs b/test/chains/epoch_block_test.exs index a17d8fb..f022d30 100644 --- a/test/chains/epoch_block_test.exs +++ b/test/chains/epoch_block_test.exs @@ -158,7 +158,7 @@ defmodule Chains.MoonbeamEpochBlockIntegrationTest do end end - defp wait_for_peak(chain, attempts \\ 30) + defp wait_for_peak(chain, attempts \\ 60) defp wait_for_peak(_chain, 0), do: flunk("timed out waiting for Moonbeam peak block from RPC") @@ -169,7 +169,7 @@ defmodule Chains.MoonbeamEpochBlockIntegrationTest do :ok _ -> - Process.sleep(1000) + Process.sleep(2_000) wait_for_peak(chain, attempts - 1) end end diff --git a/test/network/common_test.exs b/test/network/common_test.exs index e9d5ccb..127b047 100644 --- a/test/network/common_test.exs +++ b/test/network/common_test.exs @@ -68,4 +68,12 @@ defmodule Network.CommonTest do refute Map.has_key?(updated.clients, handler) refute Map.has_key?(updated.clients, key) end + + test "PeerReadyEts.read returns empty map when table is missing" do + Network.PeerReadyEts.reset() + :ets.delete(:peer_ready_connections) + assert Network.PeerReadyEts.read() == %{} + Network.PeerReadyEts.ensure() + assert :ets.info(:peer_ready_connections) != :undefined + end end diff --git a/test/network/device_notify_e2e_test.exs b/test/network/device_notify_e2e_test.exs index 839c20b..95e7c2b 100644 --- a/test/network/device_notify_e2e_test.exs +++ b/test/network/device_notify_e2e_test.exs @@ -18,10 +18,8 @@ defmodule Network.DeviceNotifyE2eTest do pid = RpcClient.connect() assert :ok = RpcClient.authenticate(pid, 1) - [ws_pid] = PubSub.subscribers({:edge, @device}) - - send( - ws_pid, + PubSub.publish( + {:edge, @device}, {:device_notify, "warning", "fleet_not_found", "Fleet is not registered on this chain"} ) diff --git a/test/network/rpc_ws_ticket_billing_test.exs b/test/network/rpc_ws_ticket_billing_test.exs index 2a56744..b764f58 100644 --- a/test/network/rpc_ws_ticket_billing_test.exs +++ b/test/network/rpc_ws_ticket_billing_test.exs @@ -7,15 +7,15 @@ defmodule Network.RpcWsTicketBillingTest do alias DiodeClient.Wallet alias Network.{RpcWsTicketBilling, TicketRequestPolicy} - @device_wallet RpcClient.clientid(1) + @device_wallet RpcClient.clientid(2) @device Wallet.address!(@device_wallet) @fleet RemoteChain.developer_fleet_address(Chains.Anvil) defp follow_up_ticket_hex do usage = TicketStore.device_usage(@device) - Edge2Client.encode_ticket_for_dio_ticket(1, - total_bytes: usage + TicketRequestPolicy.ws_usage_bytes() + Edge2Client.encode_ticket_for_dio_ticket(2, + total_bytes: usage + TicketRequestPolicy.ws_usage_bytes() + 1_000_000 ) end diff --git a/test/network/rpc_ws_ticket_request_e2e_test.exs b/test/network/rpc_ws_ticket_request_e2e_test.exs index eff5b1f..eb34a7f 100644 --- a/test/network/rpc_ws_ticket_request_e2e_test.exs +++ b/test/network/rpc_ws_ticket_request_e2e_test.exs @@ -32,8 +32,7 @@ defmodule Network.RpcWsTicketRequestE2eTest do TicketStore.increase_device_usage(@device, 1500) # Usage publish is debounced in TicketStore; deliver directly to the websocket like PubSub. - [ws_pid] = PubSub.subscribers({:edge, @device}) - send(ws_pid, {:device_usage, @device}) + PubSub.publish({:edge, @device}, {:device_usage, @device}) assert_receive {:notification, %{"method" => "dio_ticket_request", "params" => params}}, 5000 assert is_integer(params["usage"]) diff --git a/test/test_helper.exs b/test/test_helper.exs index 5663350..4cb7462 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -84,7 +84,7 @@ defmodule ChainAgent do System.put_env("WALLETS", wallets) # Need to mine one block to ensure the chain is ready (NodeProxy checks block == 0) - RemoteChain.RPC.rpc!(Chains.Anvil, "anvil_mine", [1]) + :ok = mine_anvil_block(log) IO.puts(log, "Anvil started") state else @@ -116,6 +116,24 @@ defmodule ChainAgent do Supervisor.restart_child(Diode.Supervisor, what) end + defp mine_anvil_block(log, retries \\ 10) do + url = "http://localhost:#{Chains.Anvil.port()}" + + case RemoteChain.HTTP.rpc(url, "anvil_mine", [1]) do + {:ok, _} -> + GenServer.cast(RemoteChain.NodeProxy.name(Chains.Anvil), :ensure_connections) + :ok + + {:error, _} when retries > 0 -> + IO.puts(log, "Anvil mine waiting for HTTP RPC (#{retries} retries left)") + Process.sleep(200) + mine_anvil_block(log, retries - 1) + + {:error, error} -> + raise "anvil_mine failed: #{inspect(error)}" + end + end + def handle_info({port0, {:exit_status, status}}, state = %{log: log, port: port}) do if port0 == port do IO.puts(log, "Anvil exited with status #{status}") @@ -158,11 +176,19 @@ defmodule TestHelper do def reset() do kill_clones() + Network.PeerReadyEts.ensure() + clear_device_usage_tracker() TicketStore.clear() KademliaLight.reset() restart_chain() end + defp clear_device_usage_tracker() do + if :ets.info(:device_usage_tracker) != :undefined do + :ets.delete_all_objects(:device_usage_tracker) + end + end + def set_wallets(wallets) do :persistent_term.put(:wallets, wallets) wallets @@ -187,7 +213,7 @@ defmodule TestHelper do Process.whereis(RemoteChain.Anvil) || elem(GenServer.start(ChainAgent, [], name: RemoteChain.Anvil), 1) - GenServerDbg.call(chaintask, :restart, 15_000) + GenServerDbg.call(chaintask, :restart, 30_000) end def wait(n) do From 92d04c41c9e13bf8a36f8d4ff0ca9141c4a92325 Mon Sep 17 00:00:00 2001 From: Dominic Letz Date: Mon, 13 Jul 2026 15:53:52 +0200 Subject: [PATCH 2/2] Address review feedback on ETS race, RPC errors, and nil checks. Rescue concurrent PeerReadyEts table creation, propagate any NodeProxy error tuple from RPC retry, and use is_nil/1 in chain list cache updates. --- lib/network/peer_ready_ets.ex | 7 ++++++- lib/remote_chain/chain_list.ex | 2 +- lib/remote_chain/rpc.ex | 4 ++-- 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/lib/network/peer_ready_ets.ex b/lib/network/peer_ready_ets.ex index 2e2bb9f..307ef09 100644 --- a/lib/network/peer_ready_ets.ex +++ b/lib/network/peer_ready_ets.ex @@ -17,7 +17,12 @@ defmodule Network.PeerReadyEts do def ensure do case :ets.info(@table) do :undefined -> - :ets.new(@table, [:named_table, :set, :public, read_concurrency: true]) + try do + :ets.new(@table, [:named_table, :set, :public, read_concurrency: true]) + :ok + rescue + ArgumentError -> :ok + end _ -> :ok diff --git a/lib/remote_chain/chain_list.ex b/lib/remote_chain/chain_list.ex index f09ff81..a3218ad 100644 --- a/lib/remote_chain/chain_list.ex +++ b/lib/remote_chain/chain_list.ex @@ -167,7 +167,7 @@ defmodule RemoteChain.ChainList do Enum.each(chains_by_id, fn {id, chain} -> key = cache_key(id) - if not only_cached? or Globals.get(key) != nil do + if not only_cached? or not is_nil(Globals.get(key)) do Globals.put(key, chain) end end) diff --git a/lib/remote_chain/rpc.ex b/lib/remote_chain/rpc.ex index 479ae43..889cc5c 100644 --- a/lib/remote_chain/rpc.ex +++ b/lib/remote_chain/rpc.ex @@ -72,8 +72,8 @@ defmodule RemoteChain.RPC do Process.sleep(200) rpc_with_retry(chain, method, params, retries_left - 1) - {:error, :disconnect} -> - {:error, :disconnect} + {:error, reason} -> + {:error, reason} end end