From 6e4d15f1505d835cbb8c25aeec3ed42d74ae69c6 Mon Sep 17 00:00:00 2001 From: Dominic Letz Date: Sat, 22 Aug 2026 18:14:16 +0200 Subject: [PATCH] Evict upstream connections whose RPC requests go unanswered During the Oasis Sapphire incident the primary provider (wss://sapphire.oasis.io/ws) kept streaming newHeads but stopped answering JSON-RPC requests entirely, while the fallback provider had died at the same time. Since connection health was only judged by block flow, NodeProxy kept routing requests into the zombie connection for minutes: every sapphire:getblockheader caller burned its 25s GenServer.call timeout, and peers requesting the same block queued behind the per-block :global.trans lock in RPCCache.rpc_direct, cascading into mass EdgeV2 disconnects until a manual restart. Add a watchdog that scans in-flight requests every 25s and evicts any WSConn (primary or fallback) with a request unanswered for 5x the caller timeout (125s). Eviction closes the connection, clears its leaked request entries, replies {:error, :disconnect} to orphaned callers, and schedules an ensure_connections refill so ChainList re-tests the URL. Also bump profiler to v0.4.7: warn_if_stuck custom callbacks now receive the monitored pid, so the EdgeV2 'stuck for 10000' warnings report the blocked peer's stacktrace instead of the monitor process's empty trace. --- lib/network/edge_v2.ex | 4 +- lib/remote_chain/node_proxy.ex | 63 ++++++++++++ mix.lock | 2 +- test/remote_chain/node_proxy_test.exs | 137 ++++++++++++++++++++++++++ 4 files changed, 203 insertions(+), 3 deletions(-) diff --git a/lib/network/edge_v2.ex b/lib/network/edge_v2.ex index 08db52e..fd02518 100644 --- a/lib/network/edge_v2.ex +++ b/lib/network/edge_v2.ex @@ -347,10 +347,10 @@ defmodule Network.EdgeV2 do do_handle_async_msg(msg, state) end, timeout: 10_000, - fun: fn -> + fun: fn peer -> Logger.warning( "#{label} stuck for 10000#{slow_rpc_provider_hint(msg)}\n" <> - Profiler.format_stacktrace(self()) + Profiler.format_stacktrace(peer) ) end ) diff --git a/lib/remote_chain/node_proxy.ex b/lib/remote_chain/node_proxy.ex index 61cde6c..3014a1a 100644 --- a/lib/remote_chain/node_proxy.ex +++ b/lib/remote_chain/node_proxy.ex @@ -14,6 +14,19 @@ defmodule RemoteChain.NodeProxy do @rate_limit_reconnect_ms 15_000 @security_level 1 + # A request that has gone unanswered for this multiple of the caller + # timeout marks its WSConn as unresponsive. By then (5 x 25s = 125s) + # five generations of callers have already died on their own 25s + # GenServer.call timeout, so no useful answer can still arrive: the + # provider accepted the socket data but is not servicing RPCs (zombie + # connection — the sapphire.oasis.io incident where blocks kept + # streaming while every request went unanswered for minutes). Evict + # the connection and let ChainList re-test the URL. + @watchdog_factor 5 + + # How often the watchdog scans the in-flight requests. + @watchdog_interval_ms @default_timeout + # A WSConn that has gone this many expected block intervals without a new # block is forcibly evicted, even though the socket itself is still alive. # Two consecutive ping cycles (each 2 * expected_block_intervall seconds) is @@ -46,9 +59,27 @@ defmodule RemoteChain.NodeProxy do File.mkdir_p!("logs") {:ok, log} = RotatingFile.start_link(file: "logs/#{state.chain}.log", name: nil) state = %NodeProxy{state | log: log} + schedule_watchdog() {:ok, ensure_connections(state)} end + @doc false + def watchdog_timeout_ms(), do: @watchdog_factor * @default_timeout + + @doc false + def watchdog_interval_ms do + :persistent_term.get({__MODULE__, :watchdog_interval_ms}, @watchdog_interval_ms) + end + + @doc false + def set_watchdog_interval_ms(ms) when is_integer(ms) and ms > 0 do + :persistent_term.put({__MODULE__, :watchdog_interval_ms}, ms) + end + + defp schedule_watchdog() do + Process.send_after(self(), :watchdog, watchdog_interval_ms()) + end + def rpc(chain, method, params) do GenServerDbg.call(name(chain), {:rpc, method, params}, @default_timeout) end @@ -232,6 +263,11 @@ defmodule RemoteChain.NodeProxy do {:noreply, %{state | lastblocks: lastblocks, lastblock: block_number}} end + def handle_info(:watchdog, state) do + schedule_watchdog() + {:noreply, prune_unresponsive_connections(state)} + end + def handle_info( {:DOWN, _ref, :process, down_pid, reason}, state = %{subscriptions: subs} @@ -420,6 +456,33 @@ defmodule RemoteChain.NodeProxy do } end + @doc false + def prune_unresponsive_connections(state = %NodeProxy{requests: requests, chain: chain}) do + cutoff = System.os_time(:millisecond) - watchdog_timeout_ms() + + unresponsive = + requests + |> Enum.filter(fn {_id, req} -> req.start_ms < cutoff end) + |> Enum.map(fn {_id, req} -> req.conn end) + |> Enum.uniq() + + if unresponsive == [] do + state + else + state = + Enum.reduce(unresponsive, state, fn conn, state -> + Logger.warning( + "Evicting unresponsive WSConn #{inspect(conn)} for #{inspect(chain)} " <> + "[#{conn_url(state, conn)}] (request unanswered for > #{watchdog_timeout_ms()}ms)" + ) + + close_and_remove(state, conn) + end) + + schedule_ensure_connections(state) + end + end + @doc false def prune_stale_connections( state = %NodeProxy{ diff --git a/mix.lock b/mix.lock index 5f2eca1..31534cf 100644 --- a/mix.lock +++ b/mix.lock @@ -48,7 +48,7 @@ "plug_cowboy": {:hex, :plug_cowboy, "2.7.5", "261f21b67aea8162239b2d6d3b4c31efde4daa22a20d80b19c2c0f21b34b270e", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "20884bf58a90ff5a5663420f5d2c368e9e15ed1ad5e911daf0916ea3c57f77ac"}, "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, "poison": {:hex, :poison, "6.0.0", "9bbe86722355e36ffb62c51a552719534257ba53f3271dacd20fbbd6621a583a", [:mix], [{:decimal, "~> 2.1", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "bb9064632b94775a3964642d6a78281c07b7be1319e0016e1643790704e739a2"}, - "profiler": {:git, "https://github.com/dominicletz/profiler.git", "dfc953fa3b4a709b1c20ddc2014350890ee60996", []}, + "profiler": {:git, "https://github.com/dominicletz/profiler.git", "a9fedbad37abbb868ba38e1df710a0e93037d889", []}, "pure_erlsha2": {:hex, :pure_erlsha2, "2.2.2", "02497ab4b53fec762b4317d9ad59aa05f1ad4158cb5eec48d5499b068aeaa199", [:rebar3], [], "hexpm", "5e546ed9b69b67148005d431e9e049902fe5fd6db283153526ac599762983e0a"}, "ranch": {:hex, :ranch, "2.2.0", "25528f82bc8d7c6152c57666ca99ec716510fe0925cb188172f41ce93117b1b0", [:make, :rebar3], [], "hexpm", "fa0b99a1780c80218a4197a59ea8d3bdae32fbff7e88527d7d8a4787eff4f8e7"}, "rotating_file": {:hex, :rotating_file, "0.1.2", "0a16c53ce63cee16eae03bd23db4cf488f13ad0a3916431cf895598f907ef3af", [:mix], [], "hexpm", "22636ada9ff27b5fa2d4a2cd41aaaa581e2fdeaaae2fb426b791012aa54d0f42"}, diff --git a/test/remote_chain/node_proxy_test.exs b/test/remote_chain/node_proxy_test.exs index cb91fa7..87b290c 100644 --- a/test/remote_chain/node_proxy_test.exs +++ b/test/remote_chain/node_proxy_test.exs @@ -581,6 +581,143 @@ defmodule RemoteChain.NodeProxyTest do end end + describe "prune_unresponsive_connections/1 watchdog" do + test "watchdog_timeout_ms is 5x the caller timeout (25s)" do + assert NodeProxy.watchdog_timeout_ms() == 5 * 25_000 + end + + test "evicts a connection whose request went unanswered past the watchdog timeout" do + # Zombie-provider regression (sapphire.oasis.io incident): the WS + # connection stayed alive and kept streaming newHeads, but never + # answered any RPC. Without the watchdog NodeProxy kept routing + # requests to it for minutes and every caller died on the 25s + # GenServer.call timeout. + conn = spawn(fn -> receive do: (:stop -> :ok) end) + from = {self(), make_ref()} + overdue_ms = System.os_time(:millisecond) - NodeProxy.watchdog_timeout_ms() - 1_000 + + state = %NodeProxy{ + chain: Chains.OasisSapphire, + connections: %{"wss://sapphire.example/ws" => conn}, + fallback: nil, + fallback_url: nil, + requests: %{ + 409 => %{ + from: from, + method: "eth_getBlockByNumber", + params: ["0xeacb9d", false], + start_ms: overdue_ms, + conn: conn, + ws_url: "wss://sapphire.example/ws" + } + } + } + + new_state = NodeProxy.prune_unresponsive_connections(state) + + assert new_state.connections == %{}, "zombie WSConn must be evicted" + assert new_state.requests == %{}, "dead requests on the evicted WSConn must be cleared" + + # The orphaned caller gets a disconnect reply (harmless if it has + # already timed out) instead of the entry leaking forever. + from_ref = elem(from, 1) + assert_receive {^from_ref, {:error, :disconnect}}, 1_000 + end + + test "keeps connections whose requests are still within the watchdog timeout" do + conn = spawn(fn -> receive do: (:stop -> :ok) end) + + state = %NodeProxy{ + chain: Chains.OasisSapphire, + connections: %{"wss://sapphire.example/ws" => conn}, + fallback: nil, + fallback_url: nil, + requests: %{ + 410 => %{ + from: {self(), make_ref()}, + method: "eth_getBlockByNumber", + params: ["0xeacb9e", false], + start_ms: System.os_time(:millisecond), + conn: conn, + ws_url: "wss://sapphire.example/ws" + } + } + } + + new_state = NodeProxy.prune_unresponsive_connections(state) + + assert new_state.connections == state.connections + assert new_state.requests == state.requests + + send(conn, :stop) + end + + test "evicts an unresponsive fallback connection too" do + conn = spawn(fn -> receive do: (:stop -> :ok) end) + overdue_ms = System.os_time(:millisecond) - NodeProxy.watchdog_timeout_ms() - 1_000 + + state = %NodeProxy{ + chain: Chains.OasisSapphire, + connections: %{}, + fallback: conn, + fallback_url: "wss://fallback.example/oasis/mainnet/", + requests: %{ + 411 => %{ + from: {self(), make_ref()}, + method: "eth_getBlockByNumber", + params: ["0xeacb9f", false], + start_ms: overdue_ms, + conn: conn, + ws_url: "wss://fallback.example/oasis/mainnet/" + } + } + } + + new_state = NodeProxy.prune_unresponsive_connections(state) + + assert new_state.fallback == nil + assert new_state.fallback_url == nil + assert new_state.requests == %{} + end + + test "handle_info(:watchdog) prunes and re-arms itself" do + prev = :persistent_term.get({NodeProxy, :watchdog_interval_ms}, nil) + + try do + NodeProxy.set_watchdog_interval_ms(10) + + conn = spawn(fn -> receive do: (:stop -> :ok) end) + overdue_ms = System.os_time(:millisecond) - NodeProxy.watchdog_timeout_ms() - 1_000 + + state = %NodeProxy{ + chain: Chains.OasisSapphire, + connections: %{"wss://sapphire.example/ws" => conn}, + fallback: nil, + fallback_url: nil, + requests: %{ + 412 => %{ + from: {self(), make_ref()}, + method: "eth_getBlockByNumber", + params: ["0xeacba0", false], + start_ms: overdue_ms, + conn: conn, + ws_url: "wss://sapphire.example/ws" + } + } + } + + assert {:noreply, new_state} = NodeProxy.handle_info(:watchdog, state) + assert new_state.connections == %{} + + # The watchdog must re-arm itself, otherwise a single run would + # silently disable the protection permanently. + assert_receive :watchdog, 1_000 + after + if prev, do: :persistent_term.put({NodeProxy, :watchdog_interval_ms}, prev) + end + end + end + # Shared helpers for the `prune_stale_connections/1` eviction tests. # Mark the stub as ready (so the handshake-stale check is not the one # evicting) and use a started_at that is within the handshake timeout.