Harden test suite reliability for Anvil and WebSocket tests

lib/network/peer_ready_ets.ex

@@ -14,18 +14,43 @@ defmodule Network.PeerReadyEts do
     :ets.new(@table, [:named_table, :set, :public, read_concurrency: true])
   end
 
+  def ensure do
+    case :ets.info(@table) do
+      :undefined ->
+        try do
+          :ets.new(@table, [:named_table, :set, :public, read_concurrency: true])
+          :ok
+        rescue
+          ArgumentError -> :ok
+        end
+
+      _ ->
+        :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

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 not is_nil(Globals.get(key)) 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

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, reason} ->
+        {:error, reason}
     end
   end
 

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

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

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"}
     )
 

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
 

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

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