refactor: remove dep from chain subscriber

Author: jyeshe

lib/ae_mdw/db/sync/block_index.ex

@@ -1,39 +1,51 @@
 defmodule AeMdw.Db.Sync.BlockIndex do
-  @moduledoc "fills :block table from backwards to get {height, nil} -> keyblock_hash"
+  @moduledoc """
+  Fills Model.Block table from backwards to save the key block hashes
+  """
 
-  require AeMdw.Db.Model
-
-  alias AeMdw.Db.Sync
   alias AeMdw.Db.Model
+  alias AeMdw.Node.Chain
+  alias AeMdw.Log
 
-  import AeMdw.{Sigil, Util, Db.Util}
+  require Model
 
-  @log_freq 10000
+  import AeMdw.Db.Util
+  import AeMdw.Sigil
 
-  ################################################################################
+  @log_freq 10_000
 
-  def sync(max_height \\ :safe) do
-    max_height = Sync.height(max_height)
+  @spec sync(Chain.height()) :: Chain.height()
+  def sync(max_height) when is_integer(max_height) do
+    max_height = Chain.checked_height(max_height)
     min_height = (max_kbi() || -1) + 1
 
     with true <- max_height >= min_height do
-      header = :aec_chain.get_key_header_by_height(max_height) |> ok!
-      hash = :aec_headers.hash_header(header) |> ok!
-      syncer = &sync_key_header(~t[block], &1, &2)
-      tracker = Sync.progress_logger(syncer, @log_freq, &log_msg/2)
-      :mnesia.transaction(fn -> max_height..min_height |> Enum.reduce(hash, tracker) end)
+      {:ok, header} = :aec_chain.get_key_header_by_height(max_height)
+      {:ok, hash} = :aec_headers.hash_header(header)
+
+      :mnesia.transaction(fn -> sync_range(max_height..min_height, hash) end)
     end
 
     max_kbi()
   end
 
-  def min_kbi(), do: kbi(&first/1)
+  @spec max_kbi() :: Chain.height() | nil
   def max_kbi(), do: kbi(&last/1)
 
+  @spec clear() :: :ok
   def clear(),
     do: :mnesia.clear_table(~t[block])
 
-  ################################################################################
+  #
+  # Private functions
+  #
+  defp sync_range(max_height..min_height, block_hash) do
+    Enum.reduce(max_height..min_height, block_hash, fn height, next_hash ->
+      if rem(height, @log_freq) == 0, do: Log.info("syncing block index at #{height}")
+
+      sync_key_header(~t[block], height, next_hash)
+    end)
+  end
 
   defp sync_key_header(table, height, hash) do
     {:ok, kh} = :aec_chain.get_header(hash)
@@ -50,7 +62,4 @@ defmodule AeMdw.Db.Sync.BlockIndex do
       {kbi, -1} -> kbi
     end
   end
-
-  defp log_msg(height, _hash),
-    do: "syncing block index at #{height}"
 end

lib/ae_mdw/db/sync/chain_subscriber.ex

@@ -0,0 +1,104 @@
+defmodule AeMdw.Db.Sync.ChainSubscriber do
+  @moduledoc """
+  Listens to sync chain events and spawns processes to sync until the height notified.
+  """
+  use GenServer
+
+  alias AeMdw.Db.Model
+  alias AeMdw.Db.Sync.BlockIndex
+  alias AeMdw.Db.Sync.Invalidate
+  alias AeMdw.Db.Sync.Transaction
+  alias AeMdw.Log
+  alias AeMdw.Sync.Watcher
+  alias __MODULE__, as: State
+
+  @typep state :: %State{
+           pid: pid() | nil,
+           fork: integer() | nil
+         }
+
+  defstruct [:pid, :fork]
+
+  require Model
+
+  import AeMdw.Node.Chain, only: [top_height: 0]
+
+  @verify_range_kbs 200
+
+  ################################################################################
+
+  @spec start_link([]) :: GenServer.on_start()
+  def start_link([]),
+    do: GenServer.start_link(__MODULE__, [], name: __MODULE__)
+
+  @spec init([]) :: {:ok, state(), {:continue, :start_sync}}
+  @impl GenServer
+  def init([]) do
+    :ets.delete_all_objects(:stat_sync_cache)
+    :aec_events.subscribe(:chain)
+    Watcher.notify_sync(self())
+    {:ok, %State{}, {:continue, :start_sync}}
+  end
+
+  @impl GenServer
+  def handle_continue(:start_sync, %State{pid: nil} = s),
+    do: {:noreply, spawn_action({Transaction, :sync, [safe_height()]}, s)}
+
+  @impl GenServer
+  def handle_info({:fork, height}, %State{pid: pid} = s) when is_integer(height) do
+    s = %{s | fork: fork_height(height, s.fork)}
+    {:noreply, (pid && s) || spawn_action(s)}
+  end
+
+  @impl GenServer
+  def handle_info({_, :chain, %{info: {:fork, header}}}, %State{} = s),
+    do: handle_info({:fork, :aec_headers.height(header)}, s)
+
+  @impl GenServer
+  def handle_info({_, :chain, %{info: {:generation, _}}}, %State{pid: pid} = s),
+    do: {:noreply, (pid && s) || spawn_action(s)}
+
+  @impl GenServer
+  def handle_info({pid, _act, _res}, %State{pid: pid, fork: fork} = s) when not is_nil(fork),
+    do: {:noreply, spawn_action(%{s | pid: nil})}
+
+  @impl GenServer
+  def handle_info({pid, _, _next_txi}, %State{pid: pid, fork: nil} = s) do
+    bi_max_kbi = BlockIndex.max_kbi()
+    is_synced? = bi_max_kbi == top_height()
+    next_state = %{s | pid: nil}
+    {:noreply, (is_synced? && next_state) || spawn_action(next_state)}
+  end
+
+  #
+  # Private functions
+  #
+  defp safe_height(),
+    do: max(0, top_height() - @verify_range_kbs)
+
+  defp spawn_action(%State{pid: nil, fork: nil} = s),
+    do: spawn_action({Transaction, :sync, [top_height() - 1]}, s)
+
+  defp spawn_action(%State{pid: nil, fork: height} = s) when not is_nil(height) do
+    Log.info("invalidation #{height}")
+    Invalidate.invalidate(height)
+    spawn_action({Transaction, :sync, [top_height()]}, %{s | fork: nil})
+  end
+
+  defp spawn_action({m, f, a}, %State{} = s) do
+    Log.info("sync action #{inspect(hd(a))}")
+    %{s | pid: spawn_link(fn -> run_action({m, f, a}) end)}
+  end
+
+  defp run_action({m, f, a} = action) do
+    result = apply(m, f, a)
+    send(__MODULE__, {self(), action, result})
+  end
+
+  defp fork_height(height1, height2) when is_integer(height1) do
+    case {height1, height2} do
+      {_, nil} -> height1
+      {h1, h2} -> min(h1, h2)
+    end
+  end
+end

lib/ae_mdw/db/sync/supervisor.ex

@@ -12,7 +12,7 @@ defmodule AeMdw.Db.Sync.Supervisor do
   @impl true
   def init(_args) do
     children = [
-      AeMdw.Db.Sync,
+      AeMdw.Db.Sync.ChainSubscriber,
       AeMdw.Db.Sync.ForkDetector
     ]
 

lib/ae_mdw/db/sync/sync.ex

@@ -5,9 +5,10 @@ defmodule AeMdw.Db.Sync.Transaction do
 
   alias AeMdw.Blocks
   alias AeMdw.Node, as: AE
+  alias AeMdw.Node.Chain
   alias AeMdw.Contract
   alias AeMdw.Db.Model
-  alias AeMdw.Db.Sync
+  alias AeMdw.Db.Sync.BlockIndex
   alias AeMdw.Db.Aex9AccountPresenceMutation
   alias AeMdw.Db.ContractEventsMutation
   alias AeMdw.Db.IntTransfer
@@ -19,6 +20,7 @@ defmodule AeMdw.Db.Sync.Transaction do
   alias AeMdw.Db.WriteFieldsMutation
   alias AeMdw.Db.WriteLinksMutation
   alias AeMdw.Db.WriteTxMutation
+  alias AeMdw.Log
   alias AeMdw.Mnesia
   alias AeMdw.Node
   alias AeMdw.Txs
@@ -29,15 +31,15 @@ defmodule AeMdw.Db.Sync.Transaction do
   import AeMdw.Db.Util
   import AeMdw.Util
 
-  @log_freq 1000
+  @log_freq 1_000
   @sync_cache_cleanup_freq 150_000
 
   ################################################################################
 
-  @spec sync(non_neg_integer() | :safe) :: pos_integer()
-  def sync(max_height \\ :safe) do
-    max_height = Sync.height((is_integer(max_height) && max_height + 1) || max_height)
-    bi_max_kbi = Sync.BlockIndex.sync(max_height) - 1
+  @spec sync(non_neg_integer()) :: pos_integer()
+  def sync(max_height) when is_integer(max_height) do
+    max_height = Chain.checked_height(max_height + 1)
+    bi_max_kbi = BlockIndex.sync(max_height) - 1
 
     case last(Model.Tx) do
       :"$end_of_table" ->
@@ -53,8 +55,13 @@ defmodule AeMdw.Db.Sync.Transaction do
 
   @spec sync(non_neg_integer(), non_neg_integer(), non_neg_integer()) :: pos_integer()
   def sync(from_height, to_height, txi) when from_height <= to_height do
-    tracker = Sync.progress_logger(&sync_generation/2, @log_freq, &log_msg/2)
-    next_txi = from_height..to_height |> Enum.reduce(txi, tracker)
+    next_txi =
+      Enum.reduce(from_height..to_height, txi, fn height, next_txi ->
+        if rem(height, @log_freq) == 0,
+          do: Log.info("syncing transactions at generation #{height}")
+
+        sync_generation(height, next_txi)
+      end)
 
     :mnesia.transaction(fn ->
       [succ_kb] = :mnesia.read(Model.Block, {to_height + 1, -1})
@@ -194,7 +201,4 @@ defmodule AeMdw.Db.Sync.Transaction do
 
     {txi + length(mb_txs), mbi + 1, mutations}
   end
-
-  defp log_msg(height, _ignore),
-    do: "syncing transactions at generation #{height}"
 end

lib/ae_mdw/db/sync/transaction.ex

@@ -0,0 +1,20 @@
+defmodule AeMdw.Node.Chain do
+  @moduledoc """
+  Calls to the chain Node excluding the ones related to :aec_db (for these please see `AeMdw.Node.Db`).
+  """
+  @type height :: non_neg_integer()
+
+  @spec checked_height(height()) :: height()
+  def checked_height(height) when is_integer(height) and height >= 0 do
+    top = top_height()
+
+    if height > top and top > 0,
+      do: raise(RuntimeError, message: "no such generation: #{height} (max = #{top})")
+
+    height
+  end
+
+  @spec top_height() :: height()
+  def top_height(),
+    do: :aec_headers.height(:aec_chain.top_header())
+end