Inomurko/merge unspent output bugfix to master v2

apps/omg/lib/omg/state.ex

@@ -193,6 +193,10 @@ defmodule OMG.State do
   end
 
   def handle_call({:deposit, deposits}, _from, state) do
+    if Code.ensure_loaded?(OMG.WatcherInfo.DB.EthEvent) do
+      :ok = Kernel.apply(OMG.WatcherInfo.DB.EthEvent, :insert_deposits!, [deposits])
+    end
+
     {:ok, db_updates, new_state} = Core.deposit(deposits, state)
 
     {:reply, {:ok, db_updates}, new_state}

apps/omg/test/omg/state/persistence_test.exs

@@ -23,6 +23,7 @@ defmodule OMG.State.PersistenceTest do
 
   require OMG.Utxo
 
+  alias Ecto.Adapters.SQL.Sandbox
   alias OMG.Block
   alias OMG.Eth.Configuration
   alias OMG.State.Transaction
@@ -57,6 +58,20 @@ defmodule OMG.State.PersistenceTest do
         strategy: :one_for_one
       )
 
+    Application.ensure_all_started(:postgrex)
+    Application.ensure_all_started(:spandex_ecto)
+    Application.ensure_all_started(:ecto)
+
+    {:ok, _} =
+      Supervisor.start_link(
+        [%{id: OMG.WatcherInfo.DB.Repo, start: {OMG.WatcherInfo.DB.Repo, :start_link, []}, type: :supervisor}],
+        strategy: :one_for_one,
+        name: WatcherInfo.Supervisor
+      )
+
+    :ok = Sandbox.checkout(OMG.WatcherInfo.DB.Repo)
+    Sandbox.mode(OMG.WatcherInfo.DB.Repo, {:shared, self()})
+
     on_exit(fn ->
       Application.put_env(:omg_db, :path, nil)
 
@@ -226,7 +241,8 @@ defmodule OMG.State.PersistenceTest do
         owner: owner.addr,
         currency: currency,
         amount: amount,
-        blknum: blknum
+        blknum: blknum,
+        eth_height: 1
       }
     end)
   end

apps/omg/test/omg/state_test.exs

@@ -21,6 +21,7 @@ defmodule OMG.StateTest do
 
   use OMG.DB.Fixtures
 
+  alias Ecto.Adapters.SQL.Sandbox
   alias OMG.State
   alias OMG.TestHelper
   alias OMG.Utxo
@@ -38,6 +39,20 @@ defmodule OMG.StateTest do
     # the pubsub is required, because `OMG.State` is broadcasting to the `OMG.Bus`
     {:ok, bus_apps} = Application.ensure_all_started(:omg_bus)
 
+    Application.ensure_all_started(:postgrex)
+    Application.ensure_all_started(:spandex_ecto)
+    Application.ensure_all_started(:ecto)
+
+    {:ok, _} =
+      Supervisor.start_link(
+        [%{id: OMG.WatcherInfo.DB.Repo, start: {OMG.WatcherInfo.DB.Repo, :start_link, []}, type: :supervisor}],
+        strategy: :one_for_one,
+        name: WatcherInfo.Supervisor
+      )
+
+    :ok = Sandbox.checkout(OMG.WatcherInfo.DB.Repo)
+    Sandbox.mode(OMG.WatcherInfo.DB.Repo, {:shared, self()})
+
     on_exit(fn ->
       (started_apps ++ bus_apps)
       |> Enum.reverse()
@@ -68,7 +83,19 @@ defmodule OMG.StateTest do
     fee = %{@eth => [1]}
 
     # deposits, transactions, utxo existence
-    assert {:ok, _} = State.deposit([%{owner: alice.addr, currency: @eth, amount: 10, blknum: 1}])
+    assert {:ok, _} =
+             State.deposit([
+               %{
+                 owner: alice.addr,
+                 currency: @eth,
+                 amount: 10,
+                 blknum: 1,
+                 root_chain_txhash: <<1::256>>,
+                 eth_height: 1,
+                 log_index: 0
+               }
+             ])
+
     assert true == State.utxo_exists?(Utxo.position(1, 0, 0))
 
     assert {:ok, _} = State.exec(TestHelper.create_recovered([{1, 0, 0, alice}], @eth, [{alice, 9}]), fee)

apps/omg_status/lib/omg_status/metric/event.ex

@@ -28,6 +28,7 @@ defmodule OMG.Status.Metric.Event do
     :ife_exit_finalizer,
     :in_flight_exit,
     :in_flight_exit_processor,
+    :in_flight_exit_deleted_processor,
     :piggyback,
     :piggyback_challenges_processor,
     :piggyback_processor,
@@ -54,7 +55,6 @@ defmodule OMG.Status.Metric.Event do
   :watcher_exit_processor_message_queue_len - OMG.Watcher.ExitProcessor message queue length
   :eventer_message_queue_len - OMG.Watcher.Eventer message queue length
   :db_message_queue_len - OMG.DB server implementation (OMG.DB.LevelDB.Server, or OMG.DB.RocksDB.Server,)  message queue length
-  :pending_block_queue_length - OMG.WatcherInfo.DB.PendingBlock queue length
   :write - OMG.DB KV layer has three types of actions: write, read,  multiread
   :read - OMG.DB KV layer has three types of actions: write, read,  multiread
   :multiread - OMG.DB KV layer has three types of actions: write, read,  multiread
@@ -78,7 +78,6 @@ defmodule OMG.Status.Metric.Event do
   def name(:watcher_exit_processor_message_queue_len), do: "watcher_exit_processor_message_queue_len"
   def name(:eventer_message_queue_len), do: "eventer_message_queue_len"
   def name(:db_message_queue_len), do: "db_message_queue_len"
-  def name(:pending_block_queue_length), do: "pending_block_queue_length"
   def name(:write), do: "db_write"
   def name(:read), do: "db_read"
   def name(:multiread), do: "db_multiread"
@@ -100,6 +99,7 @@ defmodule OMG.Status.Metric.Event do
   defp events_name(:exit_finalizer), do: "exit_finalizer_ethereum_events"
   defp events_name(:exit_challenger), do: "exit_challenger_ethereum_events"
   defp events_name(:in_flight_exit_processor), do: "in_flight_exit_processor_ethereum_events"
+  defp events_name(:in_flight_exit_deleted_processor), do: "in_flight_exit_deleted_processor_ethereum_events"
   defp events_name(:piggyback_processor), do: "piggyback_processor_ethereum_events"
   defp events_name(:competitor_processor), do: "competitor_processor_ethereum_events"
   defp events_name(:challenges_responds_processor), do: "challenges_responds_processor_ethereum_events"

apps/omg_watcher/lib/omg_watcher/block_getter.ex

@@ -141,10 +141,8 @@ defmodule OMG.Watcher.BlockGetter do
 
     case Core.validate_executions(tx_exec_results, block_application, state) do
       {:ok, state} ->
-        :ok =
-          {:child_chain, "block.get"}
-          |> OMG.Bus.Event.new(:block_received, block_application)
-          |> OMG.Bus.direct_local_broadcast()
+        if Code.ensure_loaded?(OMG.WatcherInfo.DB.Block),
+          do: Kernel.apply(OMG.WatcherInfo.BlockApplicator, :insert_block!, [block_application])
 
         {:noreply, state, {:continue, {:apply_block_step, :run_block_download_task, block_application}}}
 

apps/omg_watcher/lib/omg_watcher/exit_processor.ex

@@ -391,6 +391,9 @@ defmodule OMG.Watcher.ExitProcessor do
       )
       |> Enum.map(fn {:ok, result} -> result end)
 
+    if Code.ensure_loaded?(OMG.WatcherInfo.DB.EthEvent),
+      do: Kernel.apply(OMG.WatcherInfo.DB.EthEvent, :insert_exits!, [exits, :standard_exit, nil])
+
     {new_state, db_updates} = Core.new_exits(state, exit_maps, exit_contract_statuses)
     {:reply, {:ok, db_updates}, new_state}
   end
@@ -404,13 +407,17 @@ defmodule OMG.Watcher.ExitProcessor do
       end)
 
     # Prepare events data for internal bus
-    :ok =
+    events =
       exits
       |> Enum.map(fn %{input_utxos_pos: inputs} = event ->
         {event, inputs}
       end)
       |> Tools.to_bus_events_data()
-      |> publish_internal_bus_events("InFlightExitStarted")
+
+    :ok = publish_internal_bus_events(events, :InFlightExitStarted)
+
+    if Code.ensure_loaded?(OMG.WatcherInfo.DB.EthEvent),
+      do: Kernel.apply(OMG.WatcherInfo.DB.EthEvent, :insert_exits!, [events, :in_flight_exit, :InFlightExitStarted])
 
     {:ok, statuses} = Eth.RootChain.get_in_flight_exit_structs(contract_ife_ids)
 
@@ -446,10 +453,16 @@ defmodule OMG.Watcher.ExitProcessor do
     _ = if not Enum.empty?(exits), do: Logger.info("Recognized #{Enum.count(exits)} piggybacks: #{inspect(exits)}")
     {new_state, db_updates} = Core.new_piggybacks(state, exits)
 
-    :ok =
-      exits
-      |> Tools.to_bus_events_data()
-      |> publish_internal_bus_events("InFlightTxOutputPiggybacked")
+    events = Tools.to_bus_events_data(exits)
+    :ok = publish_internal_bus_events(events, :InFlightTxOutputPiggybacked)
+
+    if Code.ensure_loaded?(OMG.WatcherInfo.DB.EthEvent),
+      do:
+        Kernel.apply(
+          OMG.WatcherInfo.DB.EthEvent,
+          :insert_exits!,
+          [events, :in_flight_exit, :InFlightTxOutputPiggybacked]
+        )
 
     {:reply, {:ok, db_updates}, new_state}
   end
@@ -503,10 +516,16 @@ defmodule OMG.Watcher.ExitProcessor do
 
     {:ok, state3, db_updates} = Core.finalize_in_flight_exits(state2, finalizations, invalidities)
 
-    :ok =
-      events_with_utxos
-      |> Tools.to_bus_events_data()
-      |> publish_internal_bus_events("InFlightExitOutputWithdrawn")
+    events = Tools.to_bus_events_data(events_with_utxos)
+    :ok = publish_internal_bus_events(events, :InFlightExitOutputWithdrawn)
+
+    if Code.ensure_loaded?(OMG.WatcherInfo.DB.EthEvent),
+      do:
+        Kernel.apply(
+          OMG.WatcherInfo.DB.EthEvent,
+          :insert_exits!,
+          [events, :in_flight_exit, :InFlightExitOutputWithdrawn]
+        )
 
     {:reply, {:ok, state_db_updates ++ db_updates}, state3}
   end
@@ -709,7 +728,7 @@ defmodule OMG.Watcher.ExitProcessor do
 
   defp publish_internal_bus_events([], _), do: :ok
 
-  defp publish_internal_bus_events(events_data, topic) when is_list(events_data) and is_binary(topic) do
+  defp publish_internal_bus_events(events_data, topic) when is_list(events_data) and is_atom(topic) do
     {:watcher, topic}
     |> OMG.Bus.Event.new(:data, events_data)
     |> OMG.Bus.direct_local_broadcast()

apps/omg_watcher/test/omg_watcher/integration/monitor_test.exs

@@ -51,7 +51,7 @@ defmodule OMG.Watcher.MonitorTest do
 
   test "that a child process gets restarted after alarm is cleared" do
     child = ChildProcess.prepare_child()
-    {:ok, monitor_pid} = Monitor.start_link([Alarm, child])
+    {:ok, monitor_pid} = start_and_attach_a_child([Alarm, child])
     app_alarm = Alarm.ethereum_connection_error(__MODULE__)
 
     # the monitor is now started, we raise an alarm and kill it's child
@@ -85,7 +85,7 @@ defmodule OMG.Watcher.MonitorTest do
 
   test "that a child process does not get restarted if an alarm is cleared but it was not down" do
     child = ChildProcess.prepare_child()
-    {:ok, monitor_pid} = Monitor.start_link([Alarm, child])
+    {:ok, monitor_pid} = start_and_attach_a_child([Alarm, child])
     app_alarm = Alarm.ethereum_connection_error(__MODULE__)
     :ok = :alarm_handler.set_alarm(app_alarm)
     :erlang.trace(monitor_pid, true, [:receive])
@@ -102,6 +102,17 @@ defmodule OMG.Watcher.MonitorTest do
     assert Process.info(monitor_pid, :links) == {:links, links}
   end
 
+  defp start_and_attach_a_child(opts) do
+    case Monitor.start_link(opts) do
+      {:ok, monitor_pid} ->
+        {:ok, monitor_pid}
+
+      {:error, {{:badmatch, {:error, {:already_started, _}}}, _}} ->
+        Process.sleep(500)
+        start_and_attach_a_child(opts)
+    end
+  end
+
   defmodule ChildProcess do
     @moduledoc """
     Mocking a child process to Monitor

apps/omg_watcher_info/lib/omg_watcher_info/block_applicator.ex

@@ -0,0 +1,48 @@
+# Copyright 2019-2020 OMG Network Pte Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+defmodule OMG.WatcherInfo.BlockApplicator do
+  @moduledoc """
+  Handles new block applications from Watcher's `BlockGetter` and persists them for further processing
+  """
+
+  alias OMG.WatcherInfo.DB
+
+  @type block_application_t :: %{
+          eth_height: pos_integer(),
+          hash: binary(),
+          number: pos_integer(),
+          timestamp: pos_integer(),
+          transactions: [OMG.State.Transaction.Recovered.t()]
+        }
+
+  @doc """
+  Inserts a block along with transactions and outputs, does not break when block already exists.
+  """
+  @spec insert_block!(block_application_t()) :: :ok
+  def insert_block!(block) do
+    block
+    |> DB.Block.insert_from_block_application()
+    |> case do
+      {:ok, _} ->
+        :ok
+
+      # Ensures insert idempotency. Trying to add block with the same `blknum` that already exists takes no effect.
+      # See also [comment](https://github.com/omgnetwork/elixir-omg/pull/1769#discussion_r528700434)
+      {:error, "current_block", changeset, _explain} ->
+        [{:blknum, {_msg, [constraint: :unique, constraint_name: _name]}}] = changeset.errors()
+        :ok
+    end
+  end
+end