diff --git a/server/lib/ethui/services/anvil.ex b/server/lib/ethui/services/anvil.ex index 4ae2ad4..87ec27f 100644 --- a/server/lib/ethui/services/anvil.ex +++ b/server/lib/ethui/services/anvil.ex @@ -10,6 +10,17 @@ defmodule Ethui.Services.Anvil do alias Ethui.Stacks @idle_timeout :timer.minutes(10) + + # anvil replays its persisted state before it answers anything, and with + # historical states kept that runs into minutes on a busy chain. A slow load + # is not a failure: the boot is left running and the caller is told to come + # back, rather than killing anvil half way through reading its own state. + @boot_attempts 100 + @resume_attempts 20 + + # how long a stopping anvil gets to release its port before it is written off + # and left out of the pool + @port_free_attempts 100 @log_max_size 10_000 @type id :: pid | atom | {:via, atom, term} @@ -21,7 +32,8 @@ defmodule Ethui.Services.Anvil do slug: String.t(), hash: String.t(), anvil_opts: opts_map, - id: integer() + id: integer(), + boot_attempts: pos_integer() ] @type t :: %{ @@ -37,7 +49,9 @@ defmodule Ethui.Services.Anvil do chain_id: String.t(), # idle timer idle_timer: reference() | nil, - status: :suspended | :running | :failed, + # how long a first boot may take before the caller is told to retry + boot_attempts: pos_integer(), + status: :suspended | :starting | :running | :failed, # why the last boot attempt failed, when status is :failed error: term(), last_used: integer @@ -101,12 +115,11 @@ defmodule Ethui.Services.Anvil do Process.flag(:trap_exit, true) with {:ok, dir} <- data_dir(opts[:slug], opts[:hash]), - :ok <- File.mkdir_p!(dir), - {:ok, port} <- - Ethui.Stacks.HttpPorts.claim() do + :ok <- File.mkdir_p!(dir) do {:ok, %{ - port: port, + # claimed per boot, released when anvil lets go of it + port: nil, proc: nil, logs: :queue.new(), dir: dir, @@ -117,6 +130,7 @@ defmodule Ethui.Services.Anvil do idle_timer: nil, status: :suspended, error: nil, + boot_attempts: opts[:boot_attempts] || @boot_attempts, last_used: nil }} else @@ -125,7 +139,7 @@ defmodule Ethui.Services.Anvil do end @impl GenServer - def handle_info({:EXIT, _pid, exit_status}, %{port: port} = state) do + def handle_info({:EXIT, pid, exit_status}, %{proc: pid} = state) when not is_nil(pid) do case exit_status do # we killed it ourselves on suspend: the port stays claimed so the stack # keeps the same URL when it resumes @@ -133,16 +147,20 @@ defmodule Ethui.Services.Anvil do {:noreply, %{state | proc: nil}} 0 -> - Ethui.Stacks.HttpPorts.free(port) + release_port(state) {:stop, :normal, %{state | port: nil, proc: nil}} exit_code -> Logger.error("anvil exited with code #{inspect(exit_code)}") - Ethui.Stacks.HttpPorts.free(port) + release_port(state) {:stop, :normal, %{state | port: nil, proc: nil}} end end + # anything else linked to us - the process that started this stack, a task - + # is not anvil, and its exit says nothing about the chain + def handle_info({:EXIT, _pid, _status}, state), do: {:noreply, state} + def handle_info({:anvil_output, line}, state) do {:noreply, log_line(state, line)} end @@ -154,8 +172,9 @@ defmodule Ethui.Services.Anvil do Logger.info("Suspending #{state.slug}: #{last_used}") kill_proc(proc) + release_port(state) - {:noreply, %{state | proc: nil, status: :suspended, idle_timer: nil}} + {:noreply, %{state | proc: nil, port: nil, status: :suspended, idle_timer: nil}} end @impl GenServer @@ -182,6 +201,22 @@ defmodule Ethui.Services.Anvil do :failed -> {:reply, {:error, state.error}, state} + # a boot from an earlier request is still loading state: check on it + # briefly instead of starting a second anvil on the same port + :starting -> + case wait_until_ready(state.port, state.proc, @resume_attempts) do + :ok -> + Logger.info("#{slug} finished starting on port #{state.port}") + {:reply, :ok, %{state | status: :running, error: nil} |> touch()} + + {:error, :timeout} -> + {:reply, {:error, :starting}, state} + + {:exited, status} -> + {:error, error, state} = boot_exited(%{state | proc: nil}, status, true) + {:reply, {:error, error}, state} + end + :suspended -> Logger.info("restarting slug: #{slug}") @@ -196,8 +231,9 @@ defmodule Ethui.Services.Anvil do def handle_cast(:destroy, %{proc: proc} = state) do _ = remove_dir(state) # proc is nil while suspended - if proc, do: GenServer.stop(proc) - {:stop, :normal, state} + if proc, do: kill_proc(proc) + release_port(state) + {:stop, :normal, %{state | port: nil}} end @impl GenServer @@ -338,11 +374,19 @@ defmodule Ethui.Services.Anvil do %{state | last_used: System.system_time(:second), idle_timer: timer} end - defp wait_until_ready(port, attempts \\ 100) + defp wait_until_ready(port, proc, attempts) do + # a process that already exited will never answer, so stop polling it: the + # exit status is what says whether anvil rejected its arguments + receive do + {:EXIT, ^proc, status} -> {:exited, status} + after + 0 -> poll_until_ready(port, proc, attempts) + end + end - defp wait_until_ready(_port, 0), do: {:error, :timeout} + defp poll_until_ready(_port, _proc, 0), do: {:error, :timeout} - defp wait_until_ready(port, attempts) do + defp poll_until_ready(port, proc, attempts) do url = "http://127.0.0.1:#{port}" body = @@ -356,7 +400,9 @@ defmodule Ethui.Services.Anvil do case :httpc.request( :post, {String.to_charlist(url), [], ~c"application/json", body}, - [], + # anvil accepts the connection before it finishes loading state, and + # without these the request blocks this GenServer indefinitely + [{:timeout, 1_000}, {:connect_timeout, 500}], [{:body_format, :binary}] ) do {:ok, {{_, 200, _}, _, _}} -> @@ -364,34 +410,49 @@ defmodule Ethui.Services.Anvil do _ -> Process.sleep(100) - wait_until_ready(port, attempts - 1) + wait_until_ready(port, proc, attempts - 1) end end - defp start_anvil(%{dir: dir, chain_id: chain_id, args: args, slug: slug} = state) do - # The port is claimed once, in init/1, and held for the lifetime of the - # stack: claiming a fresh one on every resume leaked the previous one and - # moved the stack's URL out from under anyone holding it. - port = state.port - pid = self() + defp start_anvil(state, may_quarantine_state \\ true) + + # Each boot takes a fresh port. Rebinding the one the previous anvil had means + # racing its shutdown, and a stale process answering on it reads as a healthy + # boot. The port is external to nobody: callers reach a stack by its slug and + # the proxy resolves the port per request. + defp start_anvil(%{slug: slug} = state, may_quarantine) do + case Ethui.Stacks.HttpPorts.claim() do + {:ok, port} -> + state = %{state | port: port} + boot_anvil(state, anvil_args(state), may_quarantine) + {:error, reason} -> + Logger.error("No port available for #{slug}: #{inspect(reason)}") + {:error, reason, %{state | status: :suspended, error: reason}} + end + end + + defp anvil_args(%{dir: dir, chain_id: chain_id, args: args, port: port}) do # A repeated flag is a clap usage error ("cannot be used multiple times", # exit 2), not a last-one-wins override, so a caller flag that collides # with a server-managed one is dropped rather than appended to. args = drop_managed(args) - anvil_args = - args ++ - [ - "--port", - to_string(port), - "--state", - "#{dir}/state.json", - "--host", - "0.0.0.0", - "--chain-id", - to_string(chain_id) - ] ++ history_args(args) + args ++ + [ + "--port", + to_string(port), + "--state", + "#{dir}/state.json", + "--host", + "0.0.0.0", + "--chain-id", + to_string(chain_id) + ] ++ history_args(args) + end + + defp boot_anvil(%{slug: slug, port: port} = state, anvil_args, may_quarantine) do + pid = self() case MuonTrap.Daemon.start_link( anvil_bin(), @@ -399,16 +460,27 @@ defmodule Ethui.Services.Anvil do logger_fun: fn line -> send(pid, {:anvil_output, line}) end, # TODO maybe patch muontrap to have a separate stream for stderr stderr_to_stdout: true, - exit_status_to_reason: & &1 + exit_status_to_reason: & &1, + # anvil dumps its state on SIGTERM, and with historical states that + # takes longer than muontrap's 500ms default; a SIGKILL landing + # mid-dump leaves a truncated state.json that anvil then refuses to + # parse on the next boot + delay_to_sigkill: :timer.seconds(30) ) do {:ok, proc} -> - case wait_until_ready(port) do + case wait_until_ready(port, proc, state.boot_attempts) do :ok -> Logger.info("restarting slug with port: #{slug} #{port}") {:ok, %{state | proc: proc, status: :running, error: nil} |> touch()} - {:error, reason} -> - failed_to_boot(state, proc, reason) + {:exited, status} -> + boot_exited(%{state | proc: nil}, status, may_quarantine) + + # still loading: leave it alone. Killing anvil part way through + # reading its state is what leaves that state unreadable. + {:error, :timeout} -> + Logger.info("#{slug} is still starting on port #{port}") + {:error, :starting, %{state | proc: proc, status: :starting, error: :starting}} end {:error, reason} -> @@ -419,30 +491,63 @@ defmodule Ethui.Services.Anvil do # anvil writes the real reason to stdout and exits before the RPC port ever # opens - a rejected flag exits 2 immediately - so without replaying its - # output all that reaches the logs is our own readiness timeout. - defp failed_to_boot(%{slug: slug} = state, proc, reason) do + # output nothing explains the failure. + defp boot_exited(%{slug: slug} = state, status, may_quarantine) do {lines, state} = drain_output(state) - exit_status = stop_proc(proc) + release_port(state) + state = %{state | port: nil} Logger.error( - "Failed to start anvil for #{slug}: #{inspect(reason)}" <> - exit_description(exit_status) <> output_description(lines) + "Failed to start anvil for #{slug}: exited with #{inspect(status)}" <> + output_description(lines) ) - case exit_status do - # clap's usage error: anvil rejected the arguments themselves, and they - # cannot change while this process lives, so every later request fails - # from here instead of paying for another boot that cannot work - {:exited, 2} -> - {:error, {:exit, 2}, %{state | proc: nil, status: :failed, error: {:exit, 2}}} + cond do + # a truncated state file is recoverable: keep it for inspection and boot + # from scratch rather than wedging the stack forever + status == 2 and may_quarantine and state_file_rejected?(lines) -> + case quarantine_state(state) do + :ok -> start_anvil(state, false) + _ -> failed(state, {:exit, 2}) + end + + # clap's usage error on anything else: anvil rejected the arguments + # themselves, and they cannot change while this process lives, so every + # later request fails from here instead of paying for a boot that cannot + # work + status == 2 -> + failed(state, {:exit, 2}) # anything else - an unreachable fork, a port that is still bound - can # succeed on the next try, so the stack stays resumable - {:exited, code} when is_integer(code) -> - {:error, {:exit, code}, %{state | proc: nil, status: :suspended, error: {:exit, code}}} + true -> + error = {:exit, status} + {:error, error, %{state | status: :suspended, error: error}} + end + end - _ -> - {:error, reason, %{state | proc: nil, status: :suspended, error: reason}} + defp failed(state, error) do + {:error, error, %{state | status: :failed, error: error}} + end + + # "error: invalid value '/.../state.json' for '--state ': failed to + # parse json file" + defp state_file_rejected?(lines) do + Enum.any?(lines, &String.contains?(&1, "for '--state")) + end + + defp quarantine_state(%{dir: dir, slug: slug}) do + path = "#{dir}/state.json" + broken = "#{path}.corrupt" + + case File.rename(path, broken) do + :ok -> + Logger.warning("Unreadable anvil state for #{slug} moved to #{broken}") + :ok + + {:error, reason} -> + Logger.error("Failed to move unreadable anvil state for #{slug}: #{inspect(reason)}") + {:error, reason} end end @@ -458,19 +563,6 @@ defmodule Ethui.Services.Anvil do end end - defp stop_proc(proc) do - receive do - {:EXIT, ^proc, status} -> - {:exited, status} - after - 0 -> - kill_proc(proc) - :killed - end - end - - # waits for the exit so the http port is free again before the next resume - # tries to bind it defp kill_proc(proc) do Process.exit(proc, :kill) @@ -481,8 +573,48 @@ defmodule Ethui.Services.Anvil do end end - defp exit_description({:exited, status}), do: ", anvil exited with #{inspect(status)}" - defp exit_description(:killed), do: ", anvil never became ready and was killed" + # muontrap SIGTERMs anvil when the daemon stops, and anvil keeps the socket + # while it dumps state. Handing the port back before it is actually released + # would give the next stack a port it cannot bind, so one that never comes + # free is left out of the pool instead. + defp release_port(%{port: nil}), do: :ok + + defp release_port(%{port: port, slug: slug}) do + # Waiting here would block every call to this stack for as long as anvil + # takes to let go, so the pool gets the port back out of band. + spawn(fn -> + case wait_until_free(port) do + :ok -> + Ethui.Stacks.HttpPorts.free(port) + + {:error, :timeout} -> + Logger.error("Port #{port} still held after #{slug} stopped, dropping it from the pool") + end + end) + + :ok + end + + # bounded well under ensure_running's call timeout: a resume queued behind a + # slow suspend still has to get an answer + defp wait_until_free(port, attempts \\ @port_free_attempts) + + defp wait_until_free(port, 0) do + Logger.error("anvil still holds port #{port}") + {:error, :timeout} + end + + defp wait_until_free(port, attempts) do + case :gen_tcp.connect(~c"127.0.0.1", port, [:binary, active: false], 100) do + {:error, _} -> + :ok + + {:ok, socket} -> + :gen_tcp.close(socket) + Process.sleep(100) + wait_until_free(port, attempts - 1) + end + end defp output_description([]), do: ", no output" diff --git a/server/lib/ethui/stacks/http_ports.ex b/server/lib/ethui/stacks/http_ports.ex index f448867..db451af 100644 --- a/server/lib/ethui/stacks/http_ports.ex +++ b/server/lib/ethui/stacks/http_ports.ex @@ -53,16 +53,22 @@ defmodule Ethui.Stacks.HttpPorts do @impl GenServer def handle_call(:claim, _from, %{range: range, claimed: claimed} = state) do - first_free = - range |> Enum.find(fn port -> not MapSet.member?(claimed, port) end) - - case first_free do - nil -> - {:reply, {:error, :no_ports_available}, state} - - port -> - new_state = %{state | claimed: MapSet.put(claimed, port)} - {:reply, {:ok, first_free}, new_state} + # A port this pool believes is free can still be held by a process that + # outlived the stack that owned it. Handing it out anyway gives the caller + # a port it cannot bind - or worse, one where the old process answers - so + # occupied ports are marked claimed and skipped. + {port, claimed} = + Enum.reduce_while(range, {nil, claimed}, fn port, {_, claimed} -> + cond do + MapSet.member?(claimed, port) -> {:cont, {nil, claimed}} + bindable?(port) -> {:halt, {port, MapSet.put(claimed, port)}} + true -> {:cont, {nil, MapSet.put(claimed, port)}} + end + end) + + case port do + nil -> {:reply, {:error, :no_ports_available}, %{state | claimed: claimed}} + port -> {:reply, {:ok, port}, %{state | claimed: claimed}} end end @@ -80,4 +86,15 @@ defmodule Ethui.Stacks.HttpPorts do def handle_info(_, state) do {:noreply, state} end + + defp bindable?(port) do + case :gen_tcp.listen(port, [:binary, reuseaddr: true]) do + {:ok, socket} -> + :gen_tcp.close(socket) + true + + {:error, _reason} -> + false + end + end end diff --git a/server/lib/ethui/stacks/server.ex b/server/lib/ethui/stacks/server.ex index 3dfc79a..ee08978 100644 --- a/server/lib/ethui/stacks/server.ex +++ b/server/lib/ethui/stacks/server.ex @@ -70,6 +70,10 @@ defmodule Ethui.Stacks.Server do {:error, {:exit, code}} -> {:error, "Stack failed to start: anvil exited with code #{code}, check its anvil options"} + # anvil is up but still replaying its state; it will answer once it has + {:error, :starting} -> + {:error, "Stack is still starting, try again shortly"} + {:error, :timeout} -> {:error, "Stack is taking too long to start"} diff --git a/server/test/ethui/services/anvil_test.exs b/server/test/ethui/services/anvil_test.exs index c028d2a..4e18bd3 100644 --- a/server/test/ethui/services/anvil_test.exs +++ b/server/test/ethui/services/anvil_test.exs @@ -7,15 +7,38 @@ defmodule Ethui.Services.AnvilTest do setup do cleanup() + # a failed assertion skips the destroy at the end of a test, and the anvil + # it started traps exits, so without this it outlives the test and keeps + # its port bound for every test that follows + on_exit(&cleanup/0) :ok end + defp data_dir_root do + Application.get_env(:ethui, Ethui.Stacks) |> Keyword.fetch!(:data_dir_root) + end + defp cleanup do Server.list() |> Enum.each(fn slug -> Server.destroy(%Stack{slug: slug}) end) + Registry.select(Ethui.Stacks.Registry, [{{{:_, :anvil}, :"$1", :_}, [], [:"$1"]}]) + |> Enum.each(fn pid -> + if Process.alive?(pid) do + ref = Process.monitor(pid) + Anvil.destroy(pid) + # destroy is a cast: without waiting, the next test claims the port + # this anvil has not released yet + receive do + {:DOWN, ^ref, :process, ^pid, _} -> :ok + after + :timer.seconds(40) -> :ok + end + end + end) + :ok end @@ -125,6 +148,76 @@ defmodule Ethui.Services.AnvilTest do Anvil.destroy(anvil) end + test "keeps chain state across suspend and resume" do + {:ok, anvil} = Anvil.start_link(ports: HttpPorts, slug: "persist", hash: "hash", id: 1) + :ok = Anvil.ensure_running(anvil) + + client = Rpc.new_client(:http, rpc_url: Anvil.url(anvil)) + {:ok, _} = Rpc.request("anvil_mine", ["0x5"]) |> Rpc.send(client) + + {:ok, %Exth.Rpc.Response.Success{result: mined}} = + Rpc.request("eth_blockNumber", []) |> Rpc.send(client) + + # the call queues behind the suspend, which only returns once anvil has + # dumped its state and released the port + send(anvil, :suspend) + :ok = Anvil.ensure_running(anvil) + + client = Rpc.new_client(:http, rpc_url: Anvil.url(anvil)) + + assert {:ok, %Exth.Rpc.Response.Success{result: ^mined}} = + Rpc.request("eth_blockNumber", []) |> Rpc.send(client) + + Anvil.destroy(anvil) + end + + test "answers that it is still starting while anvil loads" do + bin = Path.join(System.tmp_dir!(), "slow_anvil") + File.write!(bin, "#!/bin/sh\nsleep 3\nexec anvil \"$@\"\n") + File.chmod!(bin, 0o755) + + config = Application.get_env(:ethui, Ethui.Stacks) + Application.put_env(:ethui, Ethui.Stacks, Keyword.put(config, :anvil_bin, bin)) + on_exit(fn -> Application.put_env(:ethui, Ethui.Stacks, config) end) + + {:ok, anvil} = + Anvil.start_link( + ports: HttpPorts, + slug: "slow", + hash: "hash", + id: 1, + boot_attempts: 2 + ) + + # anvil is alive and still loading, so the boot is left running + assert {:error, :starting} = Anvil.ensure_running(anvil) + + Process.sleep(3_500) + assert :ok = Anvil.ensure_running(anvil) + + client = Rpc.new_client(:http, rpc_url: Anvil.url(anvil)) + assert {:ok, _} = Rpc.request("anvil_nodeInfo", []) |> Rpc.send(client) + + Anvil.destroy(anvil) + end + + test "recovers from a state file anvil cannot parse" do + dir = Path.join([data_dir_root(), "corrupt.hash", "anvil"]) + File.mkdir_p!(dir) + # what a suspend that outran anvil's state dump leaves behind + File.write!(Path.join(dir, "state.json"), "{ truncated") + + {:ok, anvil} = Anvil.start_link(ports: HttpPorts, slug: "corrupt", hash: "hash", id: 1) + + assert :ok = Anvil.ensure_running(anvil) + assert File.exists?(Path.join(dir, "state.json.corrupt")) + + client = Rpc.new_client(:http, rpc_url: Anvil.url(anvil)) + assert {:ok, _} = Rpc.request("anvil_nodeInfo", []) |> Rpc.send(client) + + Anvil.destroy(anvil) + end + test "reports a boot failure instead of retrying it forever" do {:ok, anvil} = Anvil.start_link( @@ -147,18 +240,14 @@ defmodule Ethui.Services.AnvilTest do Anvil.destroy(anvil) end - test "keeps the same port across suspend and resume" do + test "serves again after a suspend" do {:ok, anvil} = Anvil.start_link(ports: HttpPorts, slug: "resumed", hash: "hash", id: 1) :ok = Anvil.ensure_running(anvil) - url = Anvil.url(anvil) send(anvil, :suspend) - Process.sleep(200) - :ok = Anvil.ensure_running(anvil) - assert Anvil.url(anvil) == url - client = Rpc.new_client(:http, rpc_url: url) + client = Rpc.new_client(:http, rpc_url: Anvil.url(anvil)) assert {:ok, _} = Rpc.request("anvil_nodeInfo", []) |> Rpc.send(client) Anvil.destroy(anvil)