Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 55 additions & 7 deletions lib/remote_chain/chain_list.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand All @@ -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
Expand Down
185 changes: 177 additions & 8 deletions test/remote_chain/chain_list_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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})
Expand All @@ -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
Expand All @@ -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}
Expand Down Expand Up @@ -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
15 changes: 14 additions & 1 deletion test/test_helper.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 =
Expand Down
Loading