Clean setup and shutdown of Explorer.Chain.Statistics.Server

Fixes #466

apps/explorer/lib/explorer/chain/statistics/server.ex

@@ -3,10 +3,15 @@ defmodule Explorer.Chain.Statistics.Server do
 
   use GenServer
 
+  require Logger
+
   alias Explorer.Chain.Statistics
 
   @interval 1_000
 
+  defstruct statistics: %Statistics{},
+            task: nil
+
   def child_spec(_) do
     Supervisor.Spec.worker(__MODULE__, [[refresh: true]])
   end
@@ -20,30 +25,53 @@ defmodule Explorer.Chain.Statistics.Server do
     GenServer.start_link(__MODULE__, opts, name: __MODULE__)
   end
 
+  @impl GenServer
   def init(options) when is_list(options) do
     if Keyword.get(options, :refresh, true) do
       send(self(), :refresh)
     end
 
-    {:ok, %Statistics{}}
+    {:ok, %__MODULE__{}}
+  end
+
+  @impl GenServer
+
+  def handle_info(:refresh, %__MODULE__{task: task} = state) do
+    new_state =
+      case task do
+        nil ->
+          %__MODULE__{state | task: Task.Supervisor.async_nolink(Explorer.TaskSupervisor, Statistics, :fetch, [])}
+
+        _ ->
+          state
+      end
+
+    {:noreply, new_state}
   end
 
-  def handle_info(:refresh, %Statistics{} = statistics) do
+  def handle_info({ref, %Statistics{} = statistics}, %__MODULE__{task: %Task{ref: ref}} = state) do
-    Task.start_link(fn ->
+    Process.demonitor(ref, [:flush])
-      GenServer.cast(__MODULE__, {:update, Statistics.fetch()})
+    Process.send_after(self(), :refresh, @interval)
-    end)
 
-    {:noreply, statistics}
+    {:noreply, %__MODULE__{state | statistics: statistics, task: nil}}
   end
 
-  def handle_info(_, %Statistics{} = statistics), do: {:noreply, statistics}
+  def handle_info({:DOWN, ref, :process, pid, reason}, %__MODULE__{task: %Task{pid: pid, ref: ref}} = state) do
-  def handle_call(:fetch, _, %Statistics{} = statistics), do: {:reply, statistics, statistics}
+    Logger.error(fn -> "#{inspect(Statistics)}.fetch failed and could not be cached: #{inspect(reason)}" end)
-  def handle_call(_, _, %Statistics{} = statistics), do: {:noreply, statistics}
 
-  def handle_cast({:update, %Statistics{} = statistics}, %Statistics{} = _) do
     Process.send_after(self(), :refresh, @interval)
-    {:noreply, statistics}
+
+    {:noreply, %__MODULE__{state | task: nil}}
   end
 
-  def handle_cast(_, %Statistics{} = statistics), do: {:noreply, statistics}
+  @impl GenServer
+  def handle_call(:fetch, _, %__MODULE__{statistics: %Statistics{} = statistics} = state),
+    do: {:reply, statistics, state}
+
+  @impl GenServer
+  def terminate(_, %__MODULE__{task: nil}), do: :ok
+
+  def terminate(_, %__MODULE__{task: task}) do
+    Task.shutdown(task)
+  end
 end

apps/explorer/test/explorer/chain/statistics/server_test.exs

@@ -1,5 +1,7 @@
 defmodule Explorer.Chain.Statistics.ServerTest do
-  use Explorer.DataCase
+  use Explorer.DataCase, async: false
+
+  import ExUnit.CaptureLog
 
   alias Explorer.Chain.Statistics
   alias Explorer.Chain.Statistics.Server
@@ -9,21 +11,28 @@ defmodule Explorer.Chain.Statistics.ServerTest do
 
   describe "child_spec/1" do
     test "it defines a child_spec/1 that works with supervisors" do
-      assert {:ok, _} = start_supervised(Server)
+      insert(:block)
+
+      assert {:ok, pid} = start_supervised(Server)
+
+      %Server{task: %Task{pid: pid}} = :sys.get_state(pid)
+      ref = Process.monitor(pid)
+
+      assert_receive {:DOWN, ^ref, :process, ^pid, _}
     end
   end
 
   describe "init/1" do
     test "returns a new chain when not told to refresh" do
-      {:ok, statistics} = Server.init(refresh: false)
+      empty_statistics = %Statistics{}
 
-      assert statistics == %Statistics{}
+      assert {:ok, %Server{statistics: ^empty_statistics}} = Server.init(refresh: false)
     end
 
-    test "returns a new chain when told to refresh" do
+    test "returns a new Statistics when told to refresh" do
-      {:ok, statistics} = Server.init(refresh: true)
+      empty_statistics = %Statistics{}
 
-      assert statistics == %Statistics{}
+      assert {:ok, %Server{statistics: ^empty_statistics}} = Server.init(refresh: true)
     end
 
     test "refreshes when told to refresh" do
@@ -34,57 +43,78 @@ defmodule Explorer.Chain.Statistics.ServerTest do
   end
 
   describe "handle_info/2" do
-    test "returns the original chain when sent a :refresh message" do
+    setup :state
-      original = Statistics.fetch()
+
+    test "returns the original statistics when sent a :refresh message", %{
+      state: %Server{statistics: statistics} = state
+    } do
+      assert {:noreply, %Server{statistics: ^statistics, task: task}} = Server.handle_info(:refresh, state)
 
-      assert {:noreply, ^original} = Server.handle_info(:refresh, original)
+      Task.await(task)
     end
 
-    test "launches an update when sent a :refresh message" do
+    test "launches a Statistics.fetch Task update when sent a :refresh message", %{state: state} do
-      original = Statistics.fetch()
+      assert {:noreply, %Server{task: %Task{} = task}} = Server.handle_info(:refresh, state)
-      {:ok, pid} = Server.start_link()
-      chain = Server.fetch()
-      :ok = GenServer.stop(pid)
 
-      assert original.number == chain.number
+      assert %Statistics{} = Task.await(task)
     end
 
-    test "does not reply when sent any other message" do
+    test "stores successful Task in state", %{state: state} do
-      assert {:noreply, _} = Server.handle_info(:ham, %Statistics{})
+      # so that `statistics` from Task will be different
+      insert(:block)
+
+      assert {:noreply, %Server{task: %Task{ref: ref}} = refresh_state} = Server.handle_info(:refresh, state)
+
+      assert_receive {^ref, %Statistics{} = message_statistics} = message
+
+      assert {:noreply, %Server{statistics: ^message_statistics, task: nil}} =
+               Server.handle_info(message, refresh_state)
+
+      refute message_statistics == state.statistics
+    end
+
+    test "logs crashed Task", %{state: state} do
+      assert {:noreply, %Server{task: %Task{pid: pid, ref: ref}} = refresh_state} = Server.handle_info(:refresh, state)
+
+      Process.exit(pid, :boom)
+
+      assert_receive {:DOWN, ^ref, :process, ^pid, :boom} = message
+
+      captured_log =
+        capture_log(fn ->
+          assert {:noreply, %Server{task: nil}} = Server.handle_info(message, refresh_state)
+        end)
+
+      assert captured_log =~ "Explorer.Chain.Statistics.fetch failed and could not be cached: :boom"
     end
   end
 
   describe "handle_call/3" do
     test "replies with statistics when sent a :fetch message" do
       original = Statistics.fetch()
+      state = %Server{statistics: original}
 
-      assert {:reply, _, ^original} = Server.handle_call(:fetch, self(), original)
+      assert {:reply, ^original, ^state} = Server.handle_call(:fetch, self(), state)
-    end
-
-    test "does not reply when sent any other message" do
-      assert {:noreply, _} = Server.handle_call(:ham, self(), %Statistics{})
     end
   end
 
-  describe "handle_cast/2" do
+  describe "terminate/2" do
-    test "schedules a refresh of the statistics when sent an update" do
+    setup :state
-      statistics = Statistics.fetch()
 
-      Server.handle_cast({:update, statistics}, %Statistics{})
+    test "cleans up in-progress tasks when terminated", %{state: state} do
+      assert {:noreply, %Server{task: %Task{pid: pid}} = refresh_state} = Server.handle_info(:refresh, state)
 
-      assert_receive :refresh, 2_000
+      second_ref = Process.monitor(pid)
-    end
 
-    test "returns a noreply and the new incoming chain when sent an update" do
+      Server.terminate(:boom, refresh_state)
-      original = Statistics.fetch()
 
-      assert {:noreply, ^original} = Server.handle_cast({:update, original}, %Statistics{})
+      assert_receive {:DOWN, ^second_ref, :process, ^pid, :shutdown}
     end
+  end
 
-    test "returns a noreply and the old chain when sent any other cast" do
+  defp state(_) do
-      original = Statistics.fetch()
+    {:ok, state} = Server.init([])
 
-      assert {:noreply, ^original} = Server.handle_cast(:ham, original)
+    %{state: state}
-    end
   end
 end