Evict upstream connections whose RPC requests go unanswered (Sapphire zombie-connection watchdog)

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
       )

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{

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"},

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.