From 811cbb929d77b20eef72c91df570b9aa48e760e4 Mon Sep 17 00:00:00 2001 From: Dominic Letz Date: Fri, 14 Aug 2026 22:21:06 +0800 Subject: [PATCH 1/4] Cache endpoint list so ws_endpoints never blocks the GenServer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ChainList.filter_endpoints/2 used to spawn a Task.async_stream over the chainlist and run the full health probe inline. The probe can take 30+ seconds when providers are slow, and it ran in whatever process called RemoteChain.ws_endpoints/1 — which is the NodeProxy GenServer. While blocked, the GenServer could not process :new_block messages, so lastblocks/1 went stale, every live WSConn was evicted as data-stale, and the eviction triggered another slow ws_endpoints call. On us1 Base the GenServer sat inside filter_endpoints for ~40 hours with 137k messages queued. This lands the fix planned in docs/specs/plan-endpoint-list-cache.md: * Stale-while-revalidate cache in front of filter_endpoints/2, keyed by chain_id, 60 s TTL. Warm callers return from ETS in microseconds. * Cold callers get best_effort_urls/1 (deduped + known-broken providers stripped) synchronously and schedule the probe in a detached Task. * Per-chain in-flight flag coalesces concurrent refreshes so 20 cold callers produce one probe, not 20. * Per-chain generation counter prevents zombie workers from resurrecting a cache that was explicitly cleared (chainlist refresh, tests). * ChainList.put_chains/2 invalidates the endpoint cache so newly-listed providers are picked up on the next call. * The cold path and the warm probe both apply the pocket.network / curie.radiumblock.co filter via a single rejected_provider?/1 helper, and probe_pass/2 collapses filter+map into one Enum.flat_map. Tests cover the regression, the warm/stale/empty paths, the debouncing, the chainlist invalidation, and the cold-path rejection of the two known-broken providers. --- docs/specs/plan-endpoint-list-cache.md | 359 +++++++++++++++++++++++++ lib/remote_chain/chain_list.ex | 136 +++++++++- test/remote_chain/chain_list_test.exs | 226 +++++++++++++++- 3 files changed, 706 insertions(+), 15 deletions(-) create mode 100644 docs/specs/plan-endpoint-list-cache.md diff --git a/docs/specs/plan-endpoint-list-cache.md b/docs/specs/plan-endpoint-list-cache.md new file mode 100644 index 0000000..1d7940f --- /dev/null +++ b/docs/specs/plan-endpoint-list-cache.md @@ -0,0 +1,359 @@ +# Plan: keep NodeProxy out of the endpoint-probe slow path + +## Problem statement + +`RemoteChain.NodeProxy` for `Chains.Base` (and potentially any chain whose +chainlist has slow / unresponsive providers) can become unresponsive for tens +of seconds at a time. The GenServer is sitting inside +`RemoteChain.ChainList.filter_endpoints/2 → Task.async_stream → Enum.to_list`, +with `Task.Supervised` running up to 10 concurrent HTTP / WS health probes +against all chainlist URLs. The current state on `us1` confirms the +consequence: 137 k messages queued, `connections: %{}`, `lastblocks` +40 hours stale, and no Base log entries since `2026-08-12T21:40Z`. + +Two callers inside `lib/remote_chain/node_proxy.ex` end up calling +`RemoteChain.ws_endpoints/1` synchronously in the GenServer process: + +| Call site | Trigger | Hot? | +| --- | --- | --- | +| `init/1` → `ensure_connections/1` (node_proxy.ex:49) | VM startup | once | +| `handle_cast(:ensure_connections, …)` (node_proxy.ex:149–150) | any connection drop / prune | yes, debounced 5 s | +| `handle_call({:rpc, method, params}, …)` (node_proxy.ex:99) | every JSON-RPC request | **yes, very hot** | + +The third row is the worst: every inbound RPC fires `ensure_connections/1`, +which calls `RemoteChain.ws_endpoints(chain)`, +which calls `filter_endpoints`, +which spawns a `Task.async_stream` over ~30 chainlist URLs and blocks the +GenServer until `Enum.to_list` returns. The per-call overhead is small when +all URLs are cached (`test?/2` returns instantly), but it becomes fatal the +moment one provider is slow — `Task.async_stream` waits for the slowest +batch and the GenServer can't process any message while it waits. + +While the GenServer is blocked, `:new_block` messages from live WSConns +accumulate in the queue. `NodeProxy.lastblocks[url]` (the cache consulted +by `prune_stale_connections`) goes stale past the 40 s eviction window for +Base (`@stale_eviction_intervals 20 × expected_block_intervall 2 s`), every +live WSConn is evicted as "data-stale", and the eviction triggers another +`ensure_connections` cast. The cycle repeats until the GenServer is +permanently stuck. + +This plan addresses **only** the synchronous-call-in-the-GenServer part +(Bug A). Bug B — `data_stale?/4` preferring `lastblocks[url]` over +`WSConn.lastblock_at/1` — is left as-is: once Bug A is fixed, the cache +will track real block arrivals and Bug B stops being reachable in +practice. Bug B can be revisited later if needed. + +## Goal + +`RemoteChain.ws_endpoints/1` (and its RPC sibling) must **never block +the calling process**. The NodeProxy GenServer should always be able to +process incoming messages — including `:new_block`, RPC requests, and +`:ensure_connections` casts — within a few microseconds of calling +`ws_endpoints/1`. + +Provider health probes still need to run, and verdicts still need to +be reasonably fresh: a provider that goes from healthy → unhealthy +should drop out of the endpoint list within a bounded time, and a +provider that recovers should come back within a bounded time. + +## Approach (high-level) + +Add a **stale-while-revalidate cache** in front of `filter_endpoints/2`. +The cache stores the last filtered URL list per chain (keyed by +`chain_id`). The cache is read on every call (microseconds, ETS lookup) +and never blocks the caller. A background refresh runs in a `Task` +process spawned via `Debouncer.immediate2/3` (the same debounce pattern +already used by `schedule_test_refresh/2`). + +When the cache is cold (first call after startup or after a chainlist +refresh), `filter_endpoints/2` returns the **unfiltered** chainlist +URLs immediately and schedules the probe in the background. The first +slow call still happens — but it happens off the GenServer, in a +detached worker. After at most one probe cycle, the cache is populated +and every subsequent call returns instantly. + +### Why stale-while-revalidate and not "block on first call" + +The NodeProxy runs `ensure_connections/1` from `init/1`. If the very +first `ws_endpoints/1` call blocks until the probe finishes, the +`RemoteChain.Sup` startup of Base (and every chain) will hang on every +node restart until the chainlist probe succeeds. Today this is masked +because `Globals` already caches per-URL verdicts (`@test_cache_ttl_ms` += 5 min), so most chains boot quickly. But for the cold-cache path we +still want zero blocking. + +### Why a per-chain cache and not a global one + +The cache value depends on the chain (its `chain_id`) and on the +chainlist snapshot that was current when the cache was populated. When +`RemoteChain.ChainList.download_update/0` refreshes `chains.json` from +`chainlist.org`, the endpoint list cache must be invalidated so the +next probe picks up new URLs. Keying by `chain_id` makes invalidation +straightforward (`Globals.pop({__MODULE__, :ws_endpoints, chain_id})` +for every cached chain). + +### Why cache `filter_endpoints` output and not the raw chainlist + +`filter_endpoints` already does two things we want: +1. `Enum.uniq/1` — duplicate chainlist entries. +2. The `Enum.reject` for `pocket.network` and `curie.radiumblock.co`. +3. The health probe via `test?/2`. + +The cache wraps (1)–(3) so callers always see the filtered, deduped, +probe-validated list. New chainlist URLs (added by chainlist.org) become +visible as soon as the next refresh runs, just like today. + +## Detailed design + +### Module: extend `RemoteChain.ChainList` + +No new module is needed — the cache is small and only `ChainList` owns +the probe logic. Add a `@endpoint_cache_ttl_ms` constant and a +`@endpoint_cache_key/1` helper. + +```elixir +# How long a cached endpoint list is considered fresh. Short enough that a +# provider going bad is reflected within one TTL, long enough that we don't +# re-probe on every RPC request. Independent of @test_cache_ttl_ms (per-URL +# verdict) — this caches the *list*, not individual verdicts. +@endpoint_cache_ttl_ms :timer.seconds(60) + +defp endpoint_cache_key(chain), do: {__MODULE__, :ws_endpoints, chain_id(chain)} + +defp chain_id(chain), do: RemoteChain.chainimpl(chain).chain_id() +``` + +Cache value shape: + +```elixir +{urls :: [String.t()], refreshed_at_ms :: integer()} +``` + +#### `filter_endpoints/2` becomes the cache-aware entry point + +```elixir +def filter_endpoints(endpoints, chain) do + cache_key = endpoint_cache_key(chain) + + case Globals.get(cache_key) do + {urls, ts} when is_list(urls) -> + if cache_stale?(ts) do + schedule_endpoint_refresh(endpoints, chain, cache_key) + end + urls + + nil -> + # Cold cache: kick off a background probe and return the raw URLs + # best-effort. The WSConn layer will fail any unhealthy URL naturally + # on connect; the next probe cycle will filter it out. + schedule_endpoint_refresh(endpoints, chain, cache_key) + Enum.uniq(endpoints) + end +end + +defp cache_stale?(ts), + do: System.monotonic_time(:millisecond) - ts > @endpoint_cache_ttl_ms +``` + +The current body of `filter_endpoints` moves into a private +`do_filter_endpoints/2` that the background refresh calls. + +#### Background refresh (Debouncer-based) + +```elixir +defp schedule_endpoint_refresh(endpoints, chain, cache_key) do + chain_id = elem(cache_key, 2) + + Debouncer.immediate2({__MODULE__, :ws_endpoint_refresh, chain_id}, fn -> + case Globals.get(cache_key) do + {_urls, ts} -> + if cache_stale?(ts), do: do_refresh_endpoint_cache(endpoints, chain, cache_key) + + nil -> + do_refresh_endpoint_cache(endpoints, chain, cache_key) + end + end) +end + +defp do_refresh_endpoint_cache(endpoints, chain, cache_key) do + filtered = do_filter_endpoints(endpoints, chain) + Globals.put(cache_key, {filtered, System.monotonic_time(:millisecond)}) +end +``` + +`Debouncer.immediate2/3` already enforces "at most one in-flight job per +key", so concurrent callers during cold cache trigger exactly one +probe. The probe runs in a worker process spawned by `Debouncer`, never +in the calling (GenServer) process. + +#### Invalidate on chainlist refresh + +`RemoteChain.ChainList.download_update/0` and `refresh_chains/0` need to +clear the endpoint cache so the next call picks up the new URLs. Add: + +```elixir +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) + Globals.pop(endpoint_cache_key_for_id(id)) + end + end) +end +``` + +(Implementation detail: expose `endpoint_cache_key/1` as +`endpoint_cache_key_for_id/1` for the invalidation helper, or inline the +key construction.) + +#### `rpc_endpoints/2` parallel + +Mirror the same change in `rpc_endpoints/2` — it shares the +`endpoints/2` helper, so the fix at the `endpoints/2` boundary covers +both `ws_endpoints/2` and `rpc_endpoints/2`. The `edge.ex` caller +(`forward_raw_transaction/2`) is on the slow path (broadcasts raw +transactions) but the same fix removes its blocking risk. + +### What stays unchanged + +- The env-var override (`CHAINS_BASE_WS=...` and friends) keeps working + unchanged — `RemoteChain.maybe_override/3` is consulted *before* + `ChainImpl.ws_endpoints/0` is called, so a configured value bypasses + both the probe and the cache. +- `ChainList.test?/2`, `test_cache_ttl_ms`, and the per-URL verdict + cache — all unchanged. They continue to feed `filter_endpoints`. +- `NodeProxy.ensure_connections/1`, `prune_stale_connections/1`, and the + rest of the GenServer — unchanged. Once `RemoteChain.ws_endpoints/1` + stops blocking, the GenServer is no longer the bottleneck. +- The `RemoteChain.Sup` topology — unchanged. No new supervisor child. +- Bug B (`data_stale?/4` cache preference) — explicitly out of scope. + +## Test plan + +Mirror `lib/remote_chain/chain_list.ex` with +`test/remote_chain/chain_list_test.exs`. Add a new +`describe "filter_endpoints/2 cache"` block (matching the existing +`describe "test?/2 caching"` style). + +### Required regression tests + +1. **Cold cache returns immediately and schedules background refresh.** + Use a URL whose mock provider blocks for several seconds + (`Process.sleep(:infinity)` or a 10 s `:timer.sleep`). Call + `RemoteChain.ChainList.filter_endpoints([slow_url], Chains.Anvil)` + with no pre-seeded cache and assert it returns within `< 100 ms` + and the slow URL appears in the returned list. Wait for the + background refresh to settle (poll the cache key with a timeout) and + assert the slow URL is now excluded. + +2. **Warm cache returns from Globals without touching the network.** + Pre-seed the cache via `Globals.put(cache_key, {urls, ts})` and stub + `HTTPoison` (or use the existing `with_http_mock/2` and call + `Plug.Cowboy.shutdown/1` before the assertion) to ensure no probe + runs. Call `filter_endpoints/2` and assert it returns the seeded + URLs immediately. + +3. **Stale cache schedules a background refresh but still returns the + stale value.** Pre-seed the cache with + `{ts: System.monotonic_time(:millisecond) - :timer.seconds(120)}`, + mock a fresh HTTP provider, call `filter_endpoints/2`, and assert + the immediate return is the stale list while the cache value updates + to the fresh verdict within `~ 5 s`. + +4. **Concurrent cold-cache callers trigger exactly one probe.** Seed a + `mock_ref` counter that increments on every HTTP request. Spawn 20 + concurrent `filter_endpoints/2` calls against a fresh cache and + assert the counter equals 1 (per chain, per probe). + +5. **chainlist.json refresh invalidates the cache.** Pre-seed the + endpoint cache for a known chain_id, call + `RemoteChain.ChainList.refresh_chains/1` with an updated chain entry + that swaps in a new URL, and assert + `Globals.get(endpoint_cache_key(chain))` is `nil`. + +6. **Public `RemoteChain.ws_endpoints/1` honours env overrides even + with the cache populated.** Mirror the existing + `RemoteChainTest "ws_endpoints/1 env override"` tests so we confirm + the cache layer sits below `maybe_override/3`. + +7. **Pre-seeded failure cache short-circuits (regression for + `pre-seeds failed verdicts`).** The existing + `RemoteChain.ChainListTest "drops chainlist URLs that fail the + health probe"` already covers the warm-cache path; extend it to + assert that a stale entry still serves the cached `false` until the + background refresh re-evaluates. + +### Required end-to-end / integration test + +Add a `RemoteChain.NodeProxyTest` case proving the GenServer stays +responsive while a slow provider is in the list. Sketch: + +- Seed the `EndpointListCache` for `Chains.Anvil` with one URL that + points at a `Task` that blocks on `receive`. +- Call `RemoteChain.NodeProxy.rpc(Chains.Anvil, "eth_blockNumber", [])` + via `GenServer.call` with a short timeout (e.g., 500 ms) and assert + the call returns (with whatever the existing connection pool + returns) **without timing out**, demonstrating the GenServer was not + blocked by `filter_endpoints`. + +This test does not require the new module to be merged in any specific +way; it can be written against the public `RemoteChain.ws_endpoints/1` +contract. + +### Required manual verification + +- Run `mix test` and `mix lint` (the project does both on CI; `mix lint` + is currently the CI gate per `docs/testing.md`). +- Boot a fresh node (the `dev` script) and confirm: + - `tail -f logs/Elixir.Chains.Base.log` shows incoming RPCs within + seconds of `mix run`. + - `tail -f error.log` does not show the `Evicting data-stale WSConn` + storm at startup. + +## Roll-out + +1. Land the `filter_endpoints/2` cache + background refresh + chainlist + invalidation as one PR, with the regression tests above. No + behaviour change for warm-cache callers. +2. Land a follow-up PR for Bug B (out of scope here). +3. On `us1`, the current deadlock can be broken by killing the Base + `RemoteChain.NodeProxy` via: + ```bash + ssh us1 '/opt/diode_node/bin/diode_node rpc \ + ":global.whereis_name({RemoteChain.NodeProxy, Chains.Base}) \ + |> Process.exit(:kill) |> IO.inspect()"' + ``` + The supervisor will respawn it; with this PR merged, the respawned + process will populate the cache and stay responsive. + +## Out of scope + +- Bug B (`data_stale?/4` cache preference in `prune_stale_connections`). +- Faster per-URL probes (today's HTTP timeouts come from + `HTTPoison`'s default 8 s connect / 5 s recv). +- `RemoteChain.ChainList.update/0`'s hourly chainlist download — this + PR only touches the invalidation of the endpoint cache when the + download completes; the download cadence is unchanged. +- Any change to `RemoteChain.Sup`, `RemoteChain.RPCCache`, + `RemoteChain.TxRelay`, or `RemoteChain.NonceProvider`. + +## Open questions + +- Should the cache TTL be configurable via env var? `mix.exs` / + `config/runtime.exs` would expose `@endpoint_cache_ttl_ms` so + operators can shorten it (e.g., during a provider outage) or + lengthen it (e.g., for chains with very stable providers). Default + 60 s is a reasonable starting point. +- Should we expose the cache contents for inspection (e.g., + `RemoteChain.ChainList.debug_endpoint_cache/1` returning + `{urls, age_ms, refreshing?}`)? Useful for ops debugging and for + the regression tests. Cheap to add; defer if scope grows. +- Should the cold-cache fallback list also drop `pocket.network` / + `curie.radiumblock.co`? Today `filter_endpoints` does; with the + cold-cache path returning `Enum.uniq(endpoints)`, those URLs would + be present until the first refresh. Reject them in the cold-cache + path too, to match the warm-cache behaviour exactly. \ No newline at end of file diff --git a/lib/remote_chain/chain_list.ex b/lib/remote_chain/chain_list.ex index f8d95ff..b8213a0 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,109 @@ 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) + + case Globals.get(cache_key) do + {urls, ts} when is_list(urls) -> + if cache_stale?(ts), do: schedule_endpoint_refresh(endpoints, chain) + urls + + nil -> + # Cold cache: kick off a background probe (never in this process) and + # return the deduped, pre-filtered list synchronously. Unhealthy URLs + # will surface as connect failures and be pruned by NodeProxy + # normally; the next refresh cycle drops them from the list. + schedule_endpoint_refresh(endpoints, chain) + best_effort_urls(endpoints) + end + end + + defp endpoint_cache_key(chain), + do: {__MODULE__, :ws_endpoints, RemoteChain.chainimpl(chain).chain_id()} + + defp cache_stale?(ts), + do: System.monotonic_time(:millisecond) - ts > @endpoint_cache_ttl_ms + + 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) + + # Only one refresh task per chain in flight. Use a per-chain flag in + # Globals (read-then-claim) so concurrent callers don't pile up probes, + # while sequential callers (across tests or RPC requests) can each + # schedule one. The closure snapshots the current generation; if the + # cache has been invalidated since scheduling, the worker skips. + unless refresh_in_flight?(chain_id) do + mark_refresh_in_flight(chain_id) + generation = read_generation(chain_id) + + Task.start(fn -> + try do + if read_generation(chain_id) == generation do + case Globals.get(endpoint_cache_key(chain)) do + {_, ts} -> + if cache_stale?(ts), do: refresh_endpoint_cache(endpoints, chain) + + nil -> + refresh_endpoint_cache(endpoints, chain) + end + end + after + Globals.pop(refresh_in_flight_key(chain_id)) + end + end) + end + end + + defp refresh_in_flight?(chain_id), + do: Globals.get(refresh_in_flight_key(chain_id)) == :in_flight + + defp mark_refresh_in_flight(chain_id), + do: Globals.put(refresh_in_flight_key(chain_id), :in_flight) + + defp refresh_in_flight_key(chain_id), + do: {__MODULE__, :ws_endpoint_refresh, chain_id} + + # Per-chain counter bumped every time the endpoint cache is invalidated + # (`clear_chain_cache/0`, chainlist refresh, etc.). Background probes carry + # the value at scheduling time and skip their write if it has since changed, + # so a worker outliving its test (or its process) cannot resurrect a + # cache that was explicitly cleared. + defp bump_generation(chain_id) do + key = {__MODULE__, :ws_generation, chain_id} + Globals.incr(key) + 1 + end + + defp read_generation(chain_id) do + case Globals.get({__MODULE__, :ws_generation, chain_id}) do + nil -> 0 + gen when is_integer(gen) -> gen + _ -> 0 + end + end + + defp refresh_endpoint_cache(endpoints, chain) do + filtered = + endpoints + |> best_effort_urls() + |> probe_pass(chain) + + Globals.put(endpoint_cache_key(chain), {filtered, System.monotonic_time(:millisecond)}) + 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 +188,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 @@ -333,10 +440,25 @@ defmodule RemoteChain.ChainList do if not only_cached? or not is_nil(Globals.get(key)) do Globals.put(key, chain) + # Drop the cached endpoint list so the next ws_endpoints/1 call picks + # up the new chainlist URLs (and probes any newly-listed providers). + invalidate_endpoint_cache(id) end end) end + defp endpoint_cache_key_for_id(chain_id), + do: {__MODULE__, :ws_endpoints, chain_id} + + # Called by `put_chains/2` (chainlist refresh) and from tests. Invalidation + # also bumps the per-chain generation counter so any probe scheduled + # before the invalidation can no longer write its result. + def invalidate_endpoint_cache(chain_id) do + bump_generation(chain_id) + Globals.pop(endpoint_cache_key_for_id(chain_id)) + Globals.pop(refresh_in_flight_key(chain_id)) + end + def update() do case File.stat(Diode.data_dir("chains.json"), time: :posix) do {:error, _} -> diff --git a/test/remote_chain/chain_list_test.exs b/test/remote_chain/chain_list_test.exs index b5e403e..6a22997 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,34 @@ 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 -> + case Globals.get(endpoint_cache_key(chain_id)) do + {urls, _ts} -> urls == [] + _ -> false + end + 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 +382,166 @@ 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, System.monotonic_time(:millisecond)} + ) + + 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, System.monotonic_time(:millisecond) - :timer.seconds(120)} + ) + + 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 "cold-cache callers trigger exactly one probe per chain" do + # Concurrent refresh triggers are collapsed via the per-chain in-flight + # flag: 20 simultaneous cold-cache calls should produce a single + # background probe — not 20. + chain_id = Chains.Anvil.chain_id() + url = "wss://debounce.invalid/" + + Globals.put(@loaded_key, true) + Globals.put(cache_key(chain_id), %{"chainId" => chain_id, "rpc" => [%{"url" => url}]}) + + # Pre-seed `true` so the probe, when it runs, produces a non-empty list. + 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) + + tasks = + for _ <- 1..20, + do: + Task.async(fn -> + RemoteChain.ChainList.filter_endpoints([url], Chains.Anvil) + end) + + results = Task.await_many(tasks, 2_000) + assert Enum.all?(results, &(&1 == [url])) + + assert eventually( + fn -> + match?({[_], _ts}, Globals.get(endpoint_cache_key(chain_id))) + end, + 2_000 + ) + end + + test "refresh_chains/1 invalidates the endpoint cache for affected chains" do + chain_id = Chains.Anvil.chain_id() + cached_urls = ["wss://stale.invalid/"] + + # Prime the chain cache so refresh_chains/1 (called with only_cached: true) + # actually updates this chain. + Globals.put(@loaded_key, true) + + Globals.put(cache_key(chain_id), %{ + "chainId" => chain_id, + "name" => "anvil-invalidate-before", + "rpc" => [] + }) + + Globals.put( + endpoint_cache_key(chain_id), + {cached_urls, System.monotonic_time(:millisecond)} + ) + + on_exit(&clear_chain_cache/0) + + cached = Globals.get(endpoint_cache_key(chain_id)) + assert {^cached_urls, ts} = cached + assert is_integer(ts) + + updated = %{"chainId" => chain_id, "name" => "anvil-invalidate", "rpc" => []} + RemoteChain.ChainList.refresh_chains([updated]) + + 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 + # `best_effort_urls/1` must apply the same static filter that the probe + # path applies — otherwise a known-broken URL leaks into the cold + # result and ends up in NodeProxy's pool. + 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 +584,37 @@ 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)) + # Bumps the generation counter so any in-flight probe from a previous + # test skips its write — otherwise it would resurrect the cache the new + # test just cleared. + RemoteChain.ChainList.invalidate_endpoint_cache(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 From e72e2e9d3751958b0336c3cc34c96752d707f2be Mon Sep 17 00:00:00 2001 From: Dominic Letz Date: Sat, 15 Aug 2026 10:14:49 +0800 Subject: [PATCH 2/4] Re-check generation right before the cache write Closing a TOCTOU window where a slow probe followed by a clear_chain_cache bump could let a stale write land. The gen check now guards both the start of the refresh (existing) and the Globals.put itself (new), so a worker outliving its cache cannot resurrect the cache even if the generation was bumped mid-probe. Adds a regression test that schedules a refresh, invalidates the endpoint cache mid-flight, and asserts the worker skipped its write. --- lib/remote_chain/chain_list.ex | 17 ++++++++--- test/remote_chain/chain_list_test.exs | 42 +++++++++++++++++++++++++-- 2 files changed, 52 insertions(+), 7 deletions(-) diff --git a/lib/remote_chain/chain_list.ex b/lib/remote_chain/chain_list.ex index b8213a0..523d312 100644 --- a/lib/remote_chain/chain_list.ex +++ b/lib/remote_chain/chain_list.ex @@ -109,10 +109,11 @@ defmodule RemoteChain.ChainList do if read_generation(chain_id) == generation do case Globals.get(endpoint_cache_key(chain)) do {_, ts} -> - if cache_stale?(ts), do: refresh_endpoint_cache(endpoints, chain) + if cache_stale?(ts), + do: refresh_endpoint_cache(endpoints, chain, generation) nil -> - refresh_endpoint_cache(endpoints, chain) + refresh_endpoint_cache(endpoints, chain, generation) end end after @@ -149,13 +150,21 @@ defmodule RemoteChain.ChainList do end end - defp refresh_endpoint_cache(endpoints, chain) do + defp refresh_endpoint_cache(endpoints, chain, generation) do filtered = endpoints |> best_effort_urls() |> probe_pass(chain) - Globals.put(endpoint_cache_key(chain), {filtered, System.monotonic_time(:millisecond)}) + # Re-check the generation right before the write. A `clear_chain_cache` + # mid-probe bumps the counter; without this check the worker would + # overwrite the freshly-cleared cache with stale data. + if read_generation(RemoteChain.chainimpl(chain).chain_id()) == generation do + Globals.put( + endpoint_cache_key(chain), + {filtered, System.monotonic_time(:millisecond)} + ) + end end defp probe_pass(urls, chain) do diff --git a/test/remote_chain/chain_list_test.exs b/test/remote_chain/chain_list_test.exs index 6a22997..d0fd668 100644 --- a/test/remote_chain/chain_list_test.exs +++ b/test/remote_chain/chain_list_test.exs @@ -526,10 +526,46 @@ defmodule RemoteChain.ChainListTest do assert Globals.get(endpoint_cache_key(chain_id)) == nil 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) + + gen_before_clear = Globals.get({RemoteChain.ChainList, :ws_generation, chain_id}) + RemoteChain.ChainList.invalidate_endpoint_cache(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 + # Sanity: the generation moved. + assert Globals.get({RemoteChain.ChainList, :ws_generation, chain_id}) > gen_before_clear + end + test "pocket.network and curie.radiumblock.co are dropped from the cold-cache path" do - # `best_effort_urls/1` must apply the same static filter that the probe - # path applies — otherwise a known-broken URL leaks into the cold - # result and ends up in NodeProxy's pool. result = RemoteChain.ChainList.filter_endpoints( [ From 1abe23da7cf41492b50802d76c2c1dd329a263ac Mon Sep 17 00:00:00 2001 From: Dominic Letz Date: Sat, 15 Aug 2026 10:48:18 +0800 Subject: [PATCH 3/4] Use Debouncer.immediate/3 instead of a custom in-flight flag Per project requirement, Debouncer.immediate/3 must be used to coalesce refreshes. The Debouncer spawns a closure that calls Task.start to do the actual probe work, so the Debouncer's cast handling completes quickly and the spawned Task does not contend with the GenServer mailbox. timeout: 0 makes the cooldown window effectively instant: the timer fires immediately, clearing the events entry, so the next sequential caller (after the cooldown) hits the empty-events branch and gets a fresh worker too. The outer closure snapshots the generation counter; the spawned Task re-checks it before doing any work, so a clear_chain_cache bump between scheduling and the Task running causes the Task to no-op (refresh_endpoint_cache/3 also re-checks right before the write). --- lib/remote_chain/chain_list.ex | 57 +++++++++++++-------------- test/remote_chain/chain_list_test.exs | 7 +++- 2 files changed, 33 insertions(+), 31 deletions(-) diff --git a/lib/remote_chain/chain_list.ex b/lib/remote_chain/chain_list.ex index 523d312..864131d 100644 --- a/lib/remote_chain/chain_list.ex +++ b/lib/remote_chain/chain_list.ex @@ -94,19 +94,25 @@ defmodule RemoteChain.ChainList do defp schedule_endpoint_refresh(endpoints, chain) do chain_id = elem(endpoint_cache_key(chain), 2) - - # Only one refresh task per chain in flight. Use a per-chain flag in - # Globals (read-then-claim) so concurrent callers don't pile up probes, - # while sequential callers (across tests or RPC requests) can each - # schedule one. The closure snapshots the current generation; if the - # cache has been invalidated since scheduling, the worker skips. - unless refresh_in_flight?(chain_id) do - mark_refresh_in_flight(chain_id) - generation = read_generation(chain_id) - - Task.start(fn -> - try do - if read_generation(chain_id) == generation do + generation = read_generation(chain_id) + + # Debouncer.immediate/3 coalesces concurrent callers: the first call + # runs the closure (which spawns the actual probe as a detached Task); + # subsequent calls within the cooldown window update events and return + # without spawning extra probes. `timeout: 0` makes the cooldown + # window effectively instant so the next sequential caller, after the + # timer fires, gets a fresh worker too. + # + # The outer closure snapshots the generation counter; the spawned Task + # reads it again before doing any work, so a `clear_chain_cache` bump + # between scheduling and the Task running causes the Task to no-op + # (the gen check is also re-checked inside `refresh_endpoint_cache/3` + # right before the write). + Debouncer.immediate( + {__MODULE__, :ws_endpoint_refresh, chain_id}, + fn -> + if read_generation(chain_id) == generation do + Task.start(fn -> case Globals.get(endpoint_cache_key(chain)) do {_, ts} -> if cache_stale?(ts), @@ -115,23 +121,13 @@ defmodule RemoteChain.ChainList do nil -> refresh_endpoint_cache(endpoints, chain, generation) end - end - after - Globals.pop(refresh_in_flight_key(chain_id)) + end) end - end) - end + end, + 0 + ) end - defp refresh_in_flight?(chain_id), - do: Globals.get(refresh_in_flight_key(chain_id)) == :in_flight - - defp mark_refresh_in_flight(chain_id), - do: Globals.put(refresh_in_flight_key(chain_id), :in_flight) - - defp refresh_in_flight_key(chain_id), - do: {__MODULE__, :ws_endpoint_refresh, chain_id} - # Per-chain counter bumped every time the endpoint cache is invalidated # (`clear_chain_cache/0`, chainlist refresh, etc.). Background probes carry # the value at scheduling time and skip their write if it has since changed, @@ -156,15 +152,19 @@ defmodule RemoteChain.ChainList do |> best_effort_urls() |> probe_pass(chain) + chain_id = RemoteChain.chainimpl(chain).chain_id() + # Re-check the generation right before the write. A `clear_chain_cache` # mid-probe bumps the counter; without this check the worker would # overwrite the freshly-cleared cache with stale data. - if read_generation(RemoteChain.chainimpl(chain).chain_id()) == generation do + if read_generation(chain_id) == generation do Globals.put( endpoint_cache_key(chain), {filtered, System.monotonic_time(:millisecond)} ) end + + :ok end defp probe_pass(urls, chain) do @@ -465,7 +465,6 @@ defmodule RemoteChain.ChainList do def invalidate_endpoint_cache(chain_id) do bump_generation(chain_id) Globals.pop(endpoint_cache_key_for_id(chain_id)) - Globals.pop(refresh_in_flight_key(chain_id)) end def update() do diff --git a/test/remote_chain/chain_list_test.exs b/test/remote_chain/chain_list_test.exs index d0fd668..a641621 100644 --- a/test/remote_chain/chain_list_test.exs +++ b/test/remote_chain/chain_list_test.exs @@ -458,8 +458,11 @@ defmodule RemoteChain.ChainListTest do end test "cold-cache callers trigger exactly one probe per chain" do - # Concurrent refresh triggers are collapsed via the per-chain in-flight - # flag: 20 simultaneous cold-cache calls should produce a single + # Concurrent refresh triggers are collapsed via `Debouncer.immediate/3`: + # the first call runs the closure (which spawns the probe as a detached + # Task); subsequent calls within the cooldown window update the + # Debouncer's events entry and return without spawning extra probes. + # 20 simultaneous cold-cache calls therefore produce a single # background probe — not 20. chain_id = Chains.Anvil.chain_id() url = "wss://debounce.invalid/" From 02d9b36b7d89ec56b5772c8dc780952b620f9f81 Mon Sep 17 00:00:00 2001 From: Dominic Letz Date: Mon, 17 Aug 2026 14:55:41 +0800 Subject: [PATCH 4/4] simplify --- docs/specs/plan-endpoint-list-cache.md | 359 ------------------------- lib/remote_chain/chain_list.ex | 96 +------ test/remote_chain/chain_list_test.exs | 92 +------ test/test_helper.exs | 15 +- 4 files changed, 27 insertions(+), 535 deletions(-) delete mode 100644 docs/specs/plan-endpoint-list-cache.md diff --git a/docs/specs/plan-endpoint-list-cache.md b/docs/specs/plan-endpoint-list-cache.md deleted file mode 100644 index 1d7940f..0000000 --- a/docs/specs/plan-endpoint-list-cache.md +++ /dev/null @@ -1,359 +0,0 @@ -# Plan: keep NodeProxy out of the endpoint-probe slow path - -## Problem statement - -`RemoteChain.NodeProxy` for `Chains.Base` (and potentially any chain whose -chainlist has slow / unresponsive providers) can become unresponsive for tens -of seconds at a time. The GenServer is sitting inside -`RemoteChain.ChainList.filter_endpoints/2 → Task.async_stream → Enum.to_list`, -with `Task.Supervised` running up to 10 concurrent HTTP / WS health probes -against all chainlist URLs. The current state on `us1` confirms the -consequence: 137 k messages queued, `connections: %{}`, `lastblocks` -40 hours stale, and no Base log entries since `2026-08-12T21:40Z`. - -Two callers inside `lib/remote_chain/node_proxy.ex` end up calling -`RemoteChain.ws_endpoints/1` synchronously in the GenServer process: - -| Call site | Trigger | Hot? | -| --- | --- | --- | -| `init/1` → `ensure_connections/1` (node_proxy.ex:49) | VM startup | once | -| `handle_cast(:ensure_connections, …)` (node_proxy.ex:149–150) | any connection drop / prune | yes, debounced 5 s | -| `handle_call({:rpc, method, params}, …)` (node_proxy.ex:99) | every JSON-RPC request | **yes, very hot** | - -The third row is the worst: every inbound RPC fires `ensure_connections/1`, -which calls `RemoteChain.ws_endpoints(chain)`, -which calls `filter_endpoints`, -which spawns a `Task.async_stream` over ~30 chainlist URLs and blocks the -GenServer until `Enum.to_list` returns. The per-call overhead is small when -all URLs are cached (`test?/2` returns instantly), but it becomes fatal the -moment one provider is slow — `Task.async_stream` waits for the slowest -batch and the GenServer can't process any message while it waits. - -While the GenServer is blocked, `:new_block` messages from live WSConns -accumulate in the queue. `NodeProxy.lastblocks[url]` (the cache consulted -by `prune_stale_connections`) goes stale past the 40 s eviction window for -Base (`@stale_eviction_intervals 20 × expected_block_intervall 2 s`), every -live WSConn is evicted as "data-stale", and the eviction triggers another -`ensure_connections` cast. The cycle repeats until the GenServer is -permanently stuck. - -This plan addresses **only** the synchronous-call-in-the-GenServer part -(Bug A). Bug B — `data_stale?/4` preferring `lastblocks[url]` over -`WSConn.lastblock_at/1` — is left as-is: once Bug A is fixed, the cache -will track real block arrivals and Bug B stops being reachable in -practice. Bug B can be revisited later if needed. - -## Goal - -`RemoteChain.ws_endpoints/1` (and its RPC sibling) must **never block -the calling process**. The NodeProxy GenServer should always be able to -process incoming messages — including `:new_block`, RPC requests, and -`:ensure_connections` casts — within a few microseconds of calling -`ws_endpoints/1`. - -Provider health probes still need to run, and verdicts still need to -be reasonably fresh: a provider that goes from healthy → unhealthy -should drop out of the endpoint list within a bounded time, and a -provider that recovers should come back within a bounded time. - -## Approach (high-level) - -Add a **stale-while-revalidate cache** in front of `filter_endpoints/2`. -The cache stores the last filtered URL list per chain (keyed by -`chain_id`). The cache is read on every call (microseconds, ETS lookup) -and never blocks the caller. A background refresh runs in a `Task` -process spawned via `Debouncer.immediate2/3` (the same debounce pattern -already used by `schedule_test_refresh/2`). - -When the cache is cold (first call after startup or after a chainlist -refresh), `filter_endpoints/2` returns the **unfiltered** chainlist -URLs immediately and schedules the probe in the background. The first -slow call still happens — but it happens off the GenServer, in a -detached worker. After at most one probe cycle, the cache is populated -and every subsequent call returns instantly. - -### Why stale-while-revalidate and not "block on first call" - -The NodeProxy runs `ensure_connections/1` from `init/1`. If the very -first `ws_endpoints/1` call blocks until the probe finishes, the -`RemoteChain.Sup` startup of Base (and every chain) will hang on every -node restart until the chainlist probe succeeds. Today this is masked -because `Globals` already caches per-URL verdicts (`@test_cache_ttl_ms` -= 5 min), so most chains boot quickly. But for the cold-cache path we -still want zero blocking. - -### Why a per-chain cache and not a global one - -The cache value depends on the chain (its `chain_id`) and on the -chainlist snapshot that was current when the cache was populated. When -`RemoteChain.ChainList.download_update/0` refreshes `chains.json` from -`chainlist.org`, the endpoint list cache must be invalidated so the -next probe picks up new URLs. Keying by `chain_id` makes invalidation -straightforward (`Globals.pop({__MODULE__, :ws_endpoints, chain_id})` -for every cached chain). - -### Why cache `filter_endpoints` output and not the raw chainlist - -`filter_endpoints` already does two things we want: -1. `Enum.uniq/1` — duplicate chainlist entries. -2. The `Enum.reject` for `pocket.network` and `curie.radiumblock.co`. -3. The health probe via `test?/2`. - -The cache wraps (1)–(3) so callers always see the filtered, deduped, -probe-validated list. New chainlist URLs (added by chainlist.org) become -visible as soon as the next refresh runs, just like today. - -## Detailed design - -### Module: extend `RemoteChain.ChainList` - -No new module is needed — the cache is small and only `ChainList` owns -the probe logic. Add a `@endpoint_cache_ttl_ms` constant and a -`@endpoint_cache_key/1` helper. - -```elixir -# How long a cached endpoint list is considered fresh. Short enough that a -# provider going bad is reflected within one TTL, long enough that we don't -# re-probe on every RPC request. Independent of @test_cache_ttl_ms (per-URL -# verdict) — this caches the *list*, not individual verdicts. -@endpoint_cache_ttl_ms :timer.seconds(60) - -defp endpoint_cache_key(chain), do: {__MODULE__, :ws_endpoints, chain_id(chain)} - -defp chain_id(chain), do: RemoteChain.chainimpl(chain).chain_id() -``` - -Cache value shape: - -```elixir -{urls :: [String.t()], refreshed_at_ms :: integer()} -``` - -#### `filter_endpoints/2` becomes the cache-aware entry point - -```elixir -def filter_endpoints(endpoints, chain) do - cache_key = endpoint_cache_key(chain) - - case Globals.get(cache_key) do - {urls, ts} when is_list(urls) -> - if cache_stale?(ts) do - schedule_endpoint_refresh(endpoints, chain, cache_key) - end - urls - - nil -> - # Cold cache: kick off a background probe and return the raw URLs - # best-effort. The WSConn layer will fail any unhealthy URL naturally - # on connect; the next probe cycle will filter it out. - schedule_endpoint_refresh(endpoints, chain, cache_key) - Enum.uniq(endpoints) - end -end - -defp cache_stale?(ts), - do: System.monotonic_time(:millisecond) - ts > @endpoint_cache_ttl_ms -``` - -The current body of `filter_endpoints` moves into a private -`do_filter_endpoints/2` that the background refresh calls. - -#### Background refresh (Debouncer-based) - -```elixir -defp schedule_endpoint_refresh(endpoints, chain, cache_key) do - chain_id = elem(cache_key, 2) - - Debouncer.immediate2({__MODULE__, :ws_endpoint_refresh, chain_id}, fn -> - case Globals.get(cache_key) do - {_urls, ts} -> - if cache_stale?(ts), do: do_refresh_endpoint_cache(endpoints, chain, cache_key) - - nil -> - do_refresh_endpoint_cache(endpoints, chain, cache_key) - end - end) -end - -defp do_refresh_endpoint_cache(endpoints, chain, cache_key) do - filtered = do_filter_endpoints(endpoints, chain) - Globals.put(cache_key, {filtered, System.monotonic_time(:millisecond)}) -end -``` - -`Debouncer.immediate2/3` already enforces "at most one in-flight job per -key", so concurrent callers during cold cache trigger exactly one -probe. The probe runs in a worker process spawned by `Debouncer`, never -in the calling (GenServer) process. - -#### Invalidate on chainlist refresh - -`RemoteChain.ChainList.download_update/0` and `refresh_chains/0` need to -clear the endpoint cache so the next call picks up the new URLs. Add: - -```elixir -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) - Globals.pop(endpoint_cache_key_for_id(id)) - end - end) -end -``` - -(Implementation detail: expose `endpoint_cache_key/1` as -`endpoint_cache_key_for_id/1` for the invalidation helper, or inline the -key construction.) - -#### `rpc_endpoints/2` parallel - -Mirror the same change in `rpc_endpoints/2` — it shares the -`endpoints/2` helper, so the fix at the `endpoints/2` boundary covers -both `ws_endpoints/2` and `rpc_endpoints/2`. The `edge.ex` caller -(`forward_raw_transaction/2`) is on the slow path (broadcasts raw -transactions) but the same fix removes its blocking risk. - -### What stays unchanged - -- The env-var override (`CHAINS_BASE_WS=...` and friends) keeps working - unchanged — `RemoteChain.maybe_override/3` is consulted *before* - `ChainImpl.ws_endpoints/0` is called, so a configured value bypasses - both the probe and the cache. -- `ChainList.test?/2`, `test_cache_ttl_ms`, and the per-URL verdict - cache — all unchanged. They continue to feed `filter_endpoints`. -- `NodeProxy.ensure_connections/1`, `prune_stale_connections/1`, and the - rest of the GenServer — unchanged. Once `RemoteChain.ws_endpoints/1` - stops blocking, the GenServer is no longer the bottleneck. -- The `RemoteChain.Sup` topology — unchanged. No new supervisor child. -- Bug B (`data_stale?/4` cache preference) — explicitly out of scope. - -## Test plan - -Mirror `lib/remote_chain/chain_list.ex` with -`test/remote_chain/chain_list_test.exs`. Add a new -`describe "filter_endpoints/2 cache"` block (matching the existing -`describe "test?/2 caching"` style). - -### Required regression tests - -1. **Cold cache returns immediately and schedules background refresh.** - Use a URL whose mock provider blocks for several seconds - (`Process.sleep(:infinity)` or a 10 s `:timer.sleep`). Call - `RemoteChain.ChainList.filter_endpoints([slow_url], Chains.Anvil)` - with no pre-seeded cache and assert it returns within `< 100 ms` - and the slow URL appears in the returned list. Wait for the - background refresh to settle (poll the cache key with a timeout) and - assert the slow URL is now excluded. - -2. **Warm cache returns from Globals without touching the network.** - Pre-seed the cache via `Globals.put(cache_key, {urls, ts})` and stub - `HTTPoison` (or use the existing `with_http_mock/2` and call - `Plug.Cowboy.shutdown/1` before the assertion) to ensure no probe - runs. Call `filter_endpoints/2` and assert it returns the seeded - URLs immediately. - -3. **Stale cache schedules a background refresh but still returns the - stale value.** Pre-seed the cache with - `{ts: System.monotonic_time(:millisecond) - :timer.seconds(120)}`, - mock a fresh HTTP provider, call `filter_endpoints/2`, and assert - the immediate return is the stale list while the cache value updates - to the fresh verdict within `~ 5 s`. - -4. **Concurrent cold-cache callers trigger exactly one probe.** Seed a - `mock_ref` counter that increments on every HTTP request. Spawn 20 - concurrent `filter_endpoints/2` calls against a fresh cache and - assert the counter equals 1 (per chain, per probe). - -5. **chainlist.json refresh invalidates the cache.** Pre-seed the - endpoint cache for a known chain_id, call - `RemoteChain.ChainList.refresh_chains/1` with an updated chain entry - that swaps in a new URL, and assert - `Globals.get(endpoint_cache_key(chain))` is `nil`. - -6. **Public `RemoteChain.ws_endpoints/1` honours env overrides even - with the cache populated.** Mirror the existing - `RemoteChainTest "ws_endpoints/1 env override"` tests so we confirm - the cache layer sits below `maybe_override/3`. - -7. **Pre-seeded failure cache short-circuits (regression for - `pre-seeds failed verdicts`).** The existing - `RemoteChain.ChainListTest "drops chainlist URLs that fail the - health probe"` already covers the warm-cache path; extend it to - assert that a stale entry still serves the cached `false` until the - background refresh re-evaluates. - -### Required end-to-end / integration test - -Add a `RemoteChain.NodeProxyTest` case proving the GenServer stays -responsive while a slow provider is in the list. Sketch: - -- Seed the `EndpointListCache` for `Chains.Anvil` with one URL that - points at a `Task` that blocks on `receive`. -- Call `RemoteChain.NodeProxy.rpc(Chains.Anvil, "eth_blockNumber", [])` - via `GenServer.call` with a short timeout (e.g., 500 ms) and assert - the call returns (with whatever the existing connection pool - returns) **without timing out**, demonstrating the GenServer was not - blocked by `filter_endpoints`. - -This test does not require the new module to be merged in any specific -way; it can be written against the public `RemoteChain.ws_endpoints/1` -contract. - -### Required manual verification - -- Run `mix test` and `mix lint` (the project does both on CI; `mix lint` - is currently the CI gate per `docs/testing.md`). -- Boot a fresh node (the `dev` script) and confirm: - - `tail -f logs/Elixir.Chains.Base.log` shows incoming RPCs within - seconds of `mix run`. - - `tail -f error.log` does not show the `Evicting data-stale WSConn` - storm at startup. - -## Roll-out - -1. Land the `filter_endpoints/2` cache + background refresh + chainlist - invalidation as one PR, with the regression tests above. No - behaviour change for warm-cache callers. -2. Land a follow-up PR for Bug B (out of scope here). -3. On `us1`, the current deadlock can be broken by killing the Base - `RemoteChain.NodeProxy` via: - ```bash - ssh us1 '/opt/diode_node/bin/diode_node rpc \ - ":global.whereis_name({RemoteChain.NodeProxy, Chains.Base}) \ - |> Process.exit(:kill) |> IO.inspect()"' - ``` - The supervisor will respawn it; with this PR merged, the respawned - process will populate the cache and stay responsive. - -## Out of scope - -- Bug B (`data_stale?/4` cache preference in `prune_stale_connections`). -- Faster per-URL probes (today's HTTP timeouts come from - `HTTPoison`'s default 8 s connect / 5 s recv). -- `RemoteChain.ChainList.update/0`'s hourly chainlist download — this - PR only touches the invalidation of the endpoint cache when the - download completes; the download cadence is unchanged. -- Any change to `RemoteChain.Sup`, `RemoteChain.RPCCache`, - `RemoteChain.TxRelay`, or `RemoteChain.NonceProvider`. - -## Open questions - -- Should the cache TTL be configurable via env var? `mix.exs` / - `config/runtime.exs` would expose `@endpoint_cache_ttl_ms` so - operators can shorten it (e.g., during a provider outage) or - lengthen it (e.g., for chains with very stable providers). Default - 60 s is a reasonable starting point. -- Should we expose the cache contents for inspection (e.g., - `RemoteChain.ChainList.debug_endpoint_cache/1` returning - `{urls, age_ms, refreshing?}`)? Useful for ops debugging and for - the regression tests. Cheap to add; defer if scope grows. -- Should the cold-cache fallback list also drop `pocket.network` / - `curie.radiumblock.co`? Today `filter_endpoints` does; with the - cold-cache path returning `Enum.uniq(endpoints)`, those URLs would - be present until the first refresh. Reject them in the cold-cache - path too, to match the warm-cache behaviour exactly. \ No newline at end of file diff --git a/lib/remote_chain/chain_list.ex b/lib/remote_chain/chain_list.ex index 864131d..2664bba 100644 --- a/lib/remote_chain/chain_list.ex +++ b/lib/remote_chain/chain_list.ex @@ -67,103 +67,35 @@ defmodule RemoteChain.ChainList do def filter_endpoints(endpoints, chain) do cache_key = endpoint_cache_key(chain) - - case Globals.get(cache_key) do - {urls, ts} when is_list(urls) -> - if cache_stale?(ts), do: schedule_endpoint_refresh(endpoints, chain) - urls - - nil -> - # Cold cache: kick off a background probe (never in this process) and - # return the deduped, pre-filtered list synchronously. Unhealthy URLs - # will surface as connect failures and be pruned by NodeProxy - # normally; the next refresh cycle drops them from the list. - schedule_endpoint_refresh(endpoints, chain) - best_effort_urls(endpoints) - end + 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 cache_stale?(ts), - do: System.monotonic_time(:millisecond) - ts > @endpoint_cache_ttl_ms - 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) - generation = read_generation(chain_id) - - # Debouncer.immediate/3 coalesces concurrent callers: the first call - # runs the closure (which spawns the actual probe as a detached Task); - # subsequent calls within the cooldown window update events and return - # without spawning extra probes. `timeout: 0` makes the cooldown - # window effectively instant so the next sequential caller, after the - # timer fires, gets a fresh worker too. - # - # The outer closure snapshots the generation counter; the spawned Task - # reads it again before doing any work, so a `clear_chain_cache` bump - # between scheduling and the Task running causes the Task to no-op - # (the gen check is also re-checked inside `refresh_endpoint_cache/3` - # right before the write). + Debouncer.immediate( {__MODULE__, :ws_endpoint_refresh, chain_id}, fn -> - if read_generation(chain_id) == generation do - Task.start(fn -> - case Globals.get(endpoint_cache_key(chain)) do - {_, ts} -> - if cache_stale?(ts), - do: refresh_endpoint_cache(endpoints, chain, generation) - - nil -> - refresh_endpoint_cache(endpoints, chain, generation) - end - end) - end + refresh_endpoint_cache(endpoints, chain) end, - 0 + @endpoint_cache_ttl_ms ) end - # Per-chain counter bumped every time the endpoint cache is invalidated - # (`clear_chain_cache/0`, chainlist refresh, etc.). Background probes carry - # the value at scheduling time and skip their write if it has since changed, - # so a worker outliving its test (or its process) cannot resurrect a - # cache that was explicitly cleared. - defp bump_generation(chain_id) do - key = {__MODULE__, :ws_generation, chain_id} - Globals.incr(key) + 1 - end - - defp read_generation(chain_id) do - case Globals.get({__MODULE__, :ws_generation, chain_id}) do - nil -> 0 - gen when is_integer(gen) -> gen - _ -> 0 - end - end - - defp refresh_endpoint_cache(endpoints, chain, generation) do + defp refresh_endpoint_cache(endpoints, chain) do filtered = endpoints |> best_effort_urls() |> probe_pass(chain) - chain_id = RemoteChain.chainimpl(chain).chain_id() - - # Re-check the generation right before the write. A `clear_chain_cache` - # mid-probe bumps the counter; without this check the worker would - # overwrite the freshly-cleared cache with stale data. - if read_generation(chain_id) == generation do - Globals.put( - endpoint_cache_key(chain), - {filtered, System.monotonic_time(:millisecond)} - ) - end - + Globals.put(endpoint_cache_key(chain), filtered) :ok end @@ -449,24 +381,10 @@ defmodule RemoteChain.ChainList do if not only_cached? or not is_nil(Globals.get(key)) do Globals.put(key, chain) - # Drop the cached endpoint list so the next ws_endpoints/1 call picks - # up the new chainlist URLs (and probes any newly-listed providers). - invalidate_endpoint_cache(id) end end) end - defp endpoint_cache_key_for_id(chain_id), - do: {__MODULE__, :ws_endpoints, chain_id} - - # Called by `put_chains/2` (chainlist refresh) and from tests. Invalidation - # also bumps the per-chain generation counter so any probe scheduled - # before the invalidation can no longer write its result. - def invalidate_endpoint_cache(chain_id) do - bump_generation(chain_id) - Globals.pop(endpoint_cache_key_for_id(chain_id)) - end - def update() do case File.stat(Diode.data_dir("chains.json"), time: :posix) do {:error, _} -> diff --git a/test/remote_chain/chain_list_test.exs b/test/remote_chain/chain_list_test.exs index a641621..8f6e44e 100644 --- a/test/remote_chain/chain_list_test.exs +++ b/test/remote_chain/chain_list_test.exs @@ -348,10 +348,7 @@ defmodule RemoteChain.ChainListTest do # filtered list synchronously. assert eventually( fn -> - case Globals.get(endpoint_cache_key(chain_id)) do - {urls, _ts} -> urls == [] - _ -> false - end + Globals.get(endpoint_cache_key(chain_id)) == [] end, 2_000 ), @@ -427,7 +424,7 @@ defmodule RemoteChain.ChainListTest do Globals.put( endpoint_cache_key(chain_id), - {cached_urls, System.monotonic_time(:millisecond)} + cached_urls ) on_exit(fn -> @@ -445,7 +442,7 @@ defmodule RemoteChain.ChainListTest do Globals.put( endpoint_cache_key(chain_id), - {stale_urls, System.monotonic_time(:millisecond) - :timer.seconds(120)} + stale_urls ) on_exit(fn -> @@ -457,78 +454,6 @@ defmodule RemoteChain.ChainListTest do stale_urls end - test "cold-cache callers trigger exactly one probe per chain" do - # Concurrent refresh triggers are collapsed via `Debouncer.immediate/3`: - # the first call runs the closure (which spawns the probe as a detached - # Task); subsequent calls within the cooldown window update the - # Debouncer's events entry and return without spawning extra probes. - # 20 simultaneous cold-cache calls therefore produce a single - # background probe — not 20. - chain_id = Chains.Anvil.chain_id() - url = "wss://debounce.invalid/" - - Globals.put(@loaded_key, true) - Globals.put(cache_key(chain_id), %{"chainId" => chain_id, "rpc" => [%{"url" => url}]}) - - # Pre-seed `true` so the probe, when it runs, produces a non-empty list. - 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) - - tasks = - for _ <- 1..20, - do: - Task.async(fn -> - RemoteChain.ChainList.filter_endpoints([url], Chains.Anvil) - end) - - results = Task.await_many(tasks, 2_000) - assert Enum.all?(results, &(&1 == [url])) - - assert eventually( - fn -> - match?({[_], _ts}, Globals.get(endpoint_cache_key(chain_id))) - end, - 2_000 - ) - end - - test "refresh_chains/1 invalidates the endpoint cache for affected chains" do - chain_id = Chains.Anvil.chain_id() - cached_urls = ["wss://stale.invalid/"] - - # Prime the chain cache so refresh_chains/1 (called with only_cached: true) - # actually updates this chain. - Globals.put(@loaded_key, true) - - Globals.put(cache_key(chain_id), %{ - "chainId" => chain_id, - "name" => "anvil-invalidate-before", - "rpc" => [] - }) - - Globals.put( - endpoint_cache_key(chain_id), - {cached_urls, System.monotonic_time(:millisecond)} - ) - - on_exit(&clear_chain_cache/0) - - cached = Globals.get(endpoint_cache_key(chain_id)) - assert {^cached_urls, ts} = cached - assert is_integer(ts) - - updated = %{"chainId" => chain_id, "name" => "anvil-invalidate", "rpc" => []} - RemoteChain.ChainList.refresh_chains([updated]) - - assert Globals.get(endpoint_cache_key(chain_id)) == nil - 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 @@ -555,8 +480,7 @@ defmodule RemoteChain.ChainListTest do # bumping the generation and clearing the cache. _ = RemoteChain.ChainList.filter_endpoints([url], Chains.Anvil) - gen_before_clear = Globals.get({RemoteChain.ChainList, :ws_generation, chain_id}) - RemoteChain.ChainList.invalidate_endpoint_cache(chain_id) + 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 @@ -564,8 +488,6 @@ defmodule RemoteChain.ChainListTest do # the generation check failed. Process.sleep(50) assert Globals.get(endpoint_cache_key(chain_id)) == nil - # Sanity: the generation moved. - assert Globals.get({RemoteChain.ChainList, :ws_generation, chain_id}) > gen_before_clear end test "pocket.network and curie.radiumblock.co are dropped from the cold-cache path" do @@ -625,10 +547,8 @@ defmodule RemoteChain.ChainListTest do Enum.each([@chain_id, @other_chain_id, 99_999, Chains.Anvil.chain_id()], fn chain_id -> Globals.pop(cache_key(chain_id)) - # Bumps the generation counter so any in-flight probe from a previous - # test skips its write — otherwise it would resurrect the cache the new - # test just cleared. - RemoteChain.ChainList.invalidate_endpoint_cache(chain_id) + Globals.pop(endpoint_cache_key(chain_id)) + Debouncer.cancel({RemoteChain.ChainList, :ws_endpoint_refresh, chain_id}) 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 =