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.