diff --git a/lib/remote_chain/chain_list.ex b/lib/remote_chain/chain_list.ex index f8d95ff..2664bba 100644 --- a/lib/remote_chain/chain_list.ex +++ b/lib/remote_chain/chain_list.ex @@ -8,6 +8,14 @@ defmodule RemoteChain.ChainList do # the background, so callers never block on provider probes. @test_cache_ttl_ms :timer.minutes(5) + # Stale-while-revalidate cache for the *list* of healthy providers per chain. + # Distinct from @test_cache_ttl_ms (per-URL verdict) — this caches the filtered + # set so `RemoteChain.ws_endpoints/1` never blocks the caller, even on the + # very first call after startup. Cold callers fall back to the unfiltered + # chainlist (deduped + known-broken providers stripped) and the probe runs in + # a detached Task, not in the caller's process. + @endpoint_cache_ttl_ms :timer.seconds(60) + # How far ahead of the local clock a block timestamp may be (producer skew). @max_future_skew_seconds 60 @@ -58,17 +66,50 @@ defmodule RemoteChain.ChainList do end def filter_endpoints(endpoints, chain) do - endpoints - |> Enum.uniq() - |> Task.async_stream(fn url -> {url, test?(url, chain)} end, + cache_key = endpoint_cache_key(chain) + schedule_endpoint_refresh(endpoints, chain) + Globals.get(cache_key) || best_effort_urls(endpoints) + end + + defp endpoint_cache_key(chain), + do: {__MODULE__, :ws_endpoints, RemoteChain.chainimpl(chain).chain_id()} + + defp best_effort_urls(endpoints), + do: endpoints |> Enum.uniq() |> Enum.reject(&rejected_provider?/1) + + defp schedule_endpoint_refresh(endpoints, chain) do + chain_id = elem(endpoint_cache_key(chain), 2) + + Debouncer.immediate( + {__MODULE__, :ws_endpoint_refresh, chain_id}, + fn -> + refresh_endpoint_cache(endpoints, chain) + end, + @endpoint_cache_ttl_ms + ) + end + + defp refresh_endpoint_cache(endpoints, chain) do + filtered = + endpoints + |> best_effort_urls() + |> probe_pass(chain) + + Globals.put(endpoint_cache_key(chain), filtered) + :ok + end + + defp probe_pass(urls, chain) do + urls + |> Task.async_stream( + fn url -> {url, test?(url, chain)} end, timeout: :infinity, max_concurrency: 10 ) |> Enum.to_list() - |> Enum.filter(fn {:ok, {_, result}} -> result end) - |> Enum.map(fn {:ok, {url, _}} -> url end) - |> Enum.reject(fn url -> - String.contains?(url, "pocket.network") or String.contains?(url, "curie.radiumblock.co") + |> Enum.flat_map(fn + {:ok, {url, true}} -> [url] + _ -> [] end) end @@ -88,6 +129,13 @@ defmodule RemoteChain.ChainList do end end + # Providers the chainlist sometimes lists but that we know are unusable. + # Applied both before the probe (so we don't waste a slot probing them) and + # after (so a URL that happens to pass the probe is still kept out of the + # pool). + defp rejected_provider?(url), + do: String.contains?(url, "pocket.network") or String.contains?(url, "curie.radiumblock.co") + defp expired?(tested_at) do System.monotonic_time(:millisecond) - tested_at > @test_cache_ttl_ms end diff --git a/test/remote_chain/chain_list_test.exs b/test/remote_chain/chain_list_test.exs index b5e403e..8f6e44e 100644 --- a/test/remote_chain/chain_list_test.exs +++ b/test/remote_chain/chain_list_test.exs @@ -325,7 +325,7 @@ defmodule RemoteChain.ChainListTest do "rpc" => [%{"url" => dead_rpc}, %{"url" => dead_ws}] }) - # Pre-seed failed verdicts so filter_endpoints does not open the network. + # Pre-seed failed verdicts so the background probe does not open the network. now = System.monotonic_time(:millisecond) Globals.put({RemoteChain.ChainList, :test, dead_ws}, {false, now}) Globals.put({RemoteChain.ChainList, :test, dead_rpc}, {false, now}) @@ -337,13 +337,31 @@ defmodule RemoteChain.ChainListTest do clear_chain_cache() end) - ws = RemoteChain.ChainList.ws_endpoints(Chains.Anvil, [extra_ws]) - rpc = RemoteChain.ChainList.rpc_endpoints(Chains.Anvil, [extra_ws]) - - assert ws == [extra_ws] - refute dead_ws in (ws || []) + # Cold cache: first call returns the deduped chainlist synchronously, + # without probing. The dead URLs are visible until the background + # refresh settles — this is the new contract: callers never block. + cold_ws = RemoteChain.ChainList.ws_endpoints(Chains.Anvil, [extra_ws]) + assert dead_ws in cold_ws + assert extra_ws in cold_ws + + # Background refresh drops the failed URLs and the warm cache returns the + # filtered list synchronously. + assert eventually( + fn -> + Globals.get(endpoint_cache_key(chain_id)) == [] + end, + 2_000 + ), + "background refresh did not populate the endpoint cache (got: " <> + inspect(Globals.get(endpoint_cache_key(chain_id))) <> ")" + + warm_ws = RemoteChain.ChainList.ws_endpoints(Chains.Anvil, [extra_ws]) + warm_rpc = RemoteChain.ChainList.rpc_endpoints(Chains.Anvil, [extra_ws]) + + assert warm_ws == [extra_ws] + refute dead_ws in warm_ws # WS-looking additional must land in :ws, not poison the rpc list when grouped. - assert rpc == nil or dead_rpc not in rpc + assert warm_rpc == nil or dead_rpc not in warm_rpc end test "deduplicates filtered chainlist and additional URLs" do @@ -361,6 +379,130 @@ defmodule RemoteChain.ChainListTest do assert RemoteChain.ChainList.ws_endpoints(Chains.Anvil, [url, url]) == [url] end) end + + test "cold cache returns synchronously and never opens a probe" do + # Regression for the eu1 Base deadlock (2026-08-14): filter_endpoints/2 + # was blocking the NodeProxy GenServer on a Task.async_stream of HTTP / + # WS health probes. With the cache, a slow probe is now background + # work and the caller returns immediately. + chain_id = Chains.Anvil.chain_id() + good_url = "wss://unreachable.invalid/ws-cold-good" + bad_url = "wss://unreachable.invalid/ws-cold-bad" + + Globals.put(@loaded_key, true) + + Globals.put(cache_key(chain_id), %{ + "chainId" => chain_id, + "rpc" => [%{"url" => good_url}, %{"url" => bad_url}] + }) + + # Pre-seed verdicts so the background probe does no network work. + now = System.monotonic_time(:millisecond) + Globals.put({RemoteChain.ChainList, :test, good_url}, {true, now}) + Globals.put({RemoteChain.ChainList, :test, bad_url}, {false, now}) + + on_exit(fn -> + Globals.pop(cache_key(chain_id)) + Globals.pop({RemoteChain.ChainList, :test, good_url}) + Globals.pop({RemoteChain.ChainList, :test, bad_url}) + clear_chain_cache() + end) + + started = System.monotonic_time(:millisecond) + result = RemoteChain.ChainList.filter_endpoints([good_url, bad_url], Chains.Anvil) + elapsed = System.monotonic_time(:millisecond) - started + + # Synchronous, no probe work in this process. + assert elapsed < 50 + assert good_url in result + assert bad_url in result + end + + test "warm cache returns the cached list without invoking the probe" do + chain_id = Chains.Anvil.chain_id() + cached_urls = ["wss://cached-a.invalid/", "wss://cached-b.invalid/"] + + Globals.put( + endpoint_cache_key(chain_id), + cached_urls + ) + + on_exit(fn -> + Globals.pop(endpoint_cache_key(chain_id)) + clear_chain_cache() + end) + + assert RemoteChain.ChainList.filter_endpoints(["wss://never.probed.invalid/"], Chains.Anvil) == + cached_urls + end + + test "stale cache returns the stale value and schedules a refresh" do + chain_id = Chains.Anvil.chain_id() + stale_urls = ["wss://stale.invalid/"] + + Globals.put( + endpoint_cache_key(chain_id), + stale_urls + ) + + on_exit(fn -> + Globals.pop(endpoint_cache_key(chain_id)) + clear_chain_cache() + end) + + assert RemoteChain.ChainList.filter_endpoints(["wss://never.invalid/"], Chains.Anvil) == + stale_urls + end + + test "a probe in flight at clear_chain_cache time does not resurrect the cache" do + # Regression for the eu1 Base deadlock (2026-08-14): a worker that + # outlives its test must not write to the endpoint cache after the + # cache has been cleared. The pre-write generation check protects + # against this: even if the worker is still inside `probe_pass/2` + # when `clear_chain_cache/0` bumps the counter, the write is + # skipped because the captured generation no longer matches. + chain_id = Chains.Anvil.chain_id() + url = "wss://toctou.invalid/" + + Globals.put(@loaded_key, true) + Globals.put(cache_key(chain_id), %{"chainId" => chain_id, "rpc" => [%{"url" => url}]}) + now = System.monotonic_time(:millisecond) + Globals.put({RemoteChain.ChainList, :test, url}, {true, now}) + + on_exit(fn -> + Globals.pop(cache_key(chain_id)) + Globals.pop({RemoteChain.ChainList, :test, url}) + clear_chain_cache() + end) + + # Schedule a refresh — a Task is spawned and the in-flight flag is + # set. Before the Task can complete, simulate `clear_chain_cache` by + # bumping the generation and clearing the cache. + _ = RemoteChain.ChainList.filter_endpoints([url], Chains.Anvil) + + Globals.pop(endpoint_cache_key(chain_id)) + assert Globals.get(endpoint_cache_key(chain_id)) == nil + + # Wait long enough for any in-flight probe to finish, then assert the + # cache stayed empty — the worker must have skipped its write because + # the generation check failed. + Process.sleep(50) + assert Globals.get(endpoint_cache_key(chain_id)) == nil + end + + test "pocket.network and curie.radiumblock.co are dropped from the cold-cache path" do + result = + RemoteChain.ChainList.filter_endpoints( + [ + "wss://rpc.pocket.network/abc", + "https://curie.radiumblock.co/rpc", + "https://good.invalid/rpc" + ], + Chains.Anvil + ) + + assert result == ["https://good.invalid/rpc"] + end end defp mock_ref(), do: {__MODULE__, :current_mock} @@ -403,8 +545,35 @@ defmodule RemoteChain.ChainListTest do defp clear_chain_cache() do Globals.pop(@loaded_key) - Enum.each([@chain_id, @other_chain_id, 99_999], fn chain_id -> + Enum.each([@chain_id, @other_chain_id, 99_999, Chains.Anvil.chain_id()], fn chain_id -> Globals.pop(cache_key(chain_id)) + Globals.pop(endpoint_cache_key(chain_id)) + Debouncer.cancel({RemoteChain.ChainList, :ws_endpoint_refresh, chain_id}) end) end + + defp endpoint_cache_key(chain_id), + do: {RemoteChain.ChainList, :ws_endpoints, chain_id} + + # Poll a condition until it holds or the timeout elapses. Returns the value + # of the last evaluation; callers should assert on the boolean. + defp eventually(fun, timeout_ms, interval_ms \\ 25) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + do_eventually(fun, deadline, interval_ms) + end + + defp do_eventually(fun, deadline, interval_ms) do + case fun.() do + true -> + true + + false -> + if System.monotonic_time(:millisecond) >= deadline do + false + else + Process.sleep(interval_ms) + do_eventually(fun, deadline, interval_ms) + end + end + end end diff --git a/test/test_helper.exs b/test/test_helper.exs index 4cb7462..693d299 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -24,7 +24,7 @@ defmodule ChainAgent do end defp do_restart(state = %{port: port}) do - System.cmd("killall", ["-w", "anvil"]) + kill_all_w("anvil") if port != nil do try do @@ -50,6 +50,19 @@ defmodule ChainAgent do await(%{state | port: port, out: ""}) end + defp kill_all_w(what) do + case System.cmd("killall", [what]) do + {_, 0} -> + # Means process was found and signal sent, + # will keep sending until process is gone + Process.sleep(100) + kill_all_w(what) + + {_, _} -> + :ok + end + end + defp await(state = %{port: port, out: out, log: log}) do if String.contains?(out, "Listening on") do hostport =