feat: add mutations abstraction to deal with mnesia updates (#342)

The goal of this change is to be able to isolate mnesia updates into
mutations which will all be ran in a single transaction and in the
least amount of time possible.

By making first-class mutations we can also debug easily what changes
any given block or transaction makes.

This also fixes random errors that come up inside mnesia transactions:
mnesia transactions do not work properly when exceptions are thrown
inside them, even if they are captured.

closes #331

Author: sborrazas

lib/ae_mdw/blocks.ex

@@ -13,8 +13,11 @@ defmodule AeMdw.Blocks do
 
   require Model
 
-  # This needs to be an actual type like AeMdw.Db.Tx.t()
-  @type block :: term()
+  @type height() :: non_neg_integer()
+  @type mbi() :: non_neg_integer()
+  @type time() :: non_neg_integer()
+  @type block_index() :: {height(), mbi()}
+  @type block :: map()
   @type cursor :: binary()
 
   @typep direction :: Mnesia.direction()

lib/ae_mdw/contract.ex

@@ -15,6 +15,7 @@ defmodule AeMdw.Contract do
   @tab __MODULE__
 
   @type id :: binary()
+  @type grouped_events() :: %{tx_hash() => [event()]}
 
   @typep fname() :: binary()
   @typep tx_hash() :: binary()
@@ -326,7 +327,7 @@ defmodule AeMdw.Contract do
     end)
   end
 
-  @spec get_grouped_events(micro_block()) :: %{tx_hash() => [event()]}
+  @spec get_grouped_events(micro_block()) :: grouped_events()
   def get_grouped_events(micro_block) do
     Enum.group_by(get_events(micro_block), fn {_event_name, %{tx_hash: tx_hash}} -> tx_hash end)
   end

lib/ae_mdw/db/model.ex

@@ -2,6 +2,9 @@ defmodule AeMdw.Db.Model do
   @moduledoc """
   Mnesia database model records.
   """
+  alias AeMdw.Blocks
+  alias AeMdw.Txs
+
   require Record
   require Ex2ms
 
@@ -31,6 +34,14 @@ defmodule AeMdw.Db.Model do
   @tx_defaults [index: -1, id: <<>>, block_index: {-1, -1}, time: -1]
   defrecord :tx, @tx_defaults
 
+  @type tx ::
+          record(:tx,
+            index: Txs.txi(),
+            id: Txs.tx_hash(),
+            block_index: Blocks.block_index(),
+            time: Blocks.time()
+          )
+
   # txs time index :
   #     index = {mb_time_msecs (0..), tx_index = (0...)},
   @time_defaults [index: {-1, -1}, unused: nil]

lib/ae_mdw/db/mutation.ex

@@ -0,0 +1,17 @@
+defprotocol AeMdw.Db.Mutation do
+  @moduledoc """
+  Protocol aimed at specifying all database mutations declaratively. This can be
+  a simple write to mnesia, or more complex logic involving multiple mnesia
+  queries or calls to other services.
+
+  The goal of this protocol is to do as much work as possible before starting
+  the mnesia transaction. Once the mnesia transaction is started, the mutate
+  function is called for every mutation generated.
+  """
+
+  @doc """
+  Abstracted function that performs the actual mutation into the mnesia database.
+  """
+  @spec mutate(t()) :: :ok
+  def mutate(mutation)
+end

lib/ae_mdw/db/mutations/aex9_account_presence_mutation.ex

@@ -0,0 +1,31 @@
+defmodule AeMdw.Db.Aex9AccountPresenceMutation do
+  @moduledoc """
+  Computes and derives Aex9 tokens, and stores it into the appropriate indexes.
+  """
+
+  alias AeMdw.Db.Sync
+  alias AeMdw.Blocks
+
+  defstruct [:height, :mbi]
+
+  @opaque t() :: %__MODULE__{
+            height: Blocks.height(),
+            mbi: Blocks.mbi()
+          }
+
+  @spec new(Blocks.height(), Blocks.mbi()) :: t()
+  def new(height, mbi) do
+    %__MODULE__{height: height, mbi: mbi}
+  end
+
+  @spec mutate(t()) :: :ok
+  def mutate(%__MODULE__{height: height, mbi: mbi}) do
+    Sync.Contract.aex9_derive_account_presence!({height, mbi})
+  end
+end
+
+defimpl AeMdw.Db.Mutation, for: AeMdw.Db.Aex9AccountPresenceMutation do
+  def mutate(mutation) do
+    @for.mutate(mutation)
+  end
+end

lib/ae_mdw/db/mutations/contract_events_mutation.ex

@@ -0,0 +1,36 @@
+defmodule AeMdw.Db.ContractEventsMutation do
+  @moduledoc """
+  Stores the internal contract calls by using the contract events.
+  """
+
+  alias AeMdw.Contract
+  alias AeMdw.Db.Contract, as: DBContract
+  alias AeMdw.Db.Sync
+  alias AeMdw.Txs
+
+  defstruct [:ct_pk, :events, :txi]
+
+  @typep event() :: Contract.event()
+  @opaque t() :: %__MODULE__{
+            ct_pk: DBContract.pubkey(),
+            events: [event()],
+            txi: Txs.txi()
+          }
+
+  @spec new(DBContract.pubkey(), [event()], Txs.txi()) :: t()
+  def new(ct_pk, events, txi) do
+    %__MODULE__{ct_pk: ct_pk, events: events, txi: txi}
+  end
+
+  @spec mutate(t()) :: :ok
+  def mutate(%__MODULE__{ct_pk: ct_pk, events: events, txi: txi}) do
+    ct_txi = Sync.Contract.get_txi(ct_pk)
+    Sync.Contract.events(events, txi, ct_txi)
+  end
+end
+
+defimpl AeMdw.Db.Mutation, for: AeMdw.Db.ContractEventsMutation do
+  def mutate(mutation) do
+    @for.mutate(mutation)
+  end
+end

lib/ae_mdw/db/mutations/mnesia_write_mutation.ex

@@ -0,0 +1,31 @@
+defmodule AeMdw.Db.MnesiaWriteMutation do
+  @moduledoc """
+  This is the most basic kind of transaction, it just inserts a record in a
+  mnesia table.
+  """
+
+  alias AeMdw.Mnesia
+
+  defstruct [:table, :record]
+
+  @opaque t() :: %__MODULE__{
+            table: Mnesia.table(),
+            record: Mnesia.record()
+          }
+
+  @spec new(Mnesia.table(), Mnesia.record()) :: t()
+  def new(table, record) do
+    %__MODULE__{table: table, record: record}
+  end
+
+  @spec mutate(t()) :: :ok
+  def mutate(%__MODULE__{table: table, record: record}) do
+    Mnesia.write(table, record)
+  end
+end
+
+defimpl AeMdw.Db.Mutation, for: AeMdw.Db.MnesiaWriteMutation do
+  def mutate(mutation) do
+    @for.mutate(mutation)
+  end
+end

lib/ae_mdw/db/mutations/write_fields_mutation.ex

@@ -0,0 +1,68 @@
+defmodule AeMdw.Db.WriteFieldsMutation do
+  @moduledoc """
+  Stores the indexes for the Fields table.
+  """
+
+  alias AeMdw.Blocks
+  alias AeMdw.Node
+  alias AeMdw.Db.Model
+  alias AeMdw.Txs
+
+  require Model
+
+  defstruct [:type, :tx, :block_index, :txi]
+
+  @opaque t() :: %__MODULE__{
+            type: Node.tx_type(),
+            tx: Model.tx(),
+            block_index: Blocks.block_index(),
+            txi: Txs.txi()
+          }
+
+  @spec new(Node.tx_type(), Model.tx(), Blocks.block_index(), Txs.txi()) :: t()
+  def new(type, tx, block_index, txi) do
+    %__MODULE__{
+      type: type,
+      tx: tx,
+      block_index: block_index,
+      txi: txi
+    }
+  end
+
+  @spec mutate(t()) :: :ok
+  def mutate(%__MODULE__{type: tx_type, tx: tx, block_index: block_index, txi: txi}) do
+    tx_type
+    |> Node.tx_ids()
+    |> Enum.each(fn {field, pos} ->
+      <<_::256>> = pk = resolve_pubkey(elem(tx, pos), tx_type, field, block_index)
+      write_field(tx_type, pos, pk, txi)
+    end)
+  end
+
+  defp write_field(tx_type, pos, pubkey, txi) do
+    m_field = Model.field(index: {tx_type, pos, pubkey, txi})
+    :mnesia.write(Model.Field, m_field, :write)
+    Model.incr_count({tx_type, pos, pubkey})
+  end
+
+  defp resolve_pubkey(id, :spend_tx, :recipient_id, block_index) do
+    case :aeser_id.specialize(id) do
+      {:name, name_hash} ->
+        AeMdw.Db.Name.ptr_resolve!(block_index, name_hash, "account_pubkey")
+
+      {_tag, pk} ->
+        pk
+    end
+  end
+
+  defp resolve_pubkey(id, _type, _field, _block_index) do
+    {_tag, pk} = :aeser_id.specialize(id)
+    pk
+  end
+end
+
+defimpl AeMdw.Db.Mutation, for: AeMdw.Db.WriteFieldsMutation do
+  def mutate(mutation) do
+    @for.mutate(mutation)
+  end
+end

lib/ae_mdw/db/mutations/write_links_mutation.ex

@@ -0,0 +1,143 @@
+defmodule AeMdw.Db.WriteLinksMutation do
+  @moduledoc """
+  Creates all the necessary indexes for both Oracles and Names depending on the
+  transaction type.
+  """
+
+  alias AeMdw.Blocks
+  alias AeMdw.Db.Model
+  alias AeMdw.Db.Sync
+  alias AeMdw.Node
+  alias AeMdw.Txs
+
+  require Model
+
+  defstruct [:type, :tx, :signed_tx, :txi, :tx_hash, :block_index]
+
+  @opaque t() :: %__MODULE__{
+            type: Node.tx_type(),
+            tx: Model.tx(),
+            signed_tx: Node.signed_tx(),
+            txi: Txs.txi(),
+            tx_hash: Txs.tx_hash(),
+            block_index: Blocks.block_index()
+          }
+
+  @spec new(
+          Node.tx_type(),
+          Model.tx(),
+          Node.signed_tx(),
+          Txs.txi(),
+          Txs.tx_hash(),
+          Blocks.block_index()
+        ) :: t()
+  def new(type, tx, signed_tx, txi, tx_hash, block_index) do
+    %__MODULE__{
+      type: type,
+      tx: tx,
+      signed_tx: signed_tx,
+      txi: txi,
+      tx_hash: tx_hash,
+      block_index: block_index
+    }
+  end
+
+  @spec mutate(t()) :: :ok
+  def mutate(%__MODULE__{
+        type: :contract_create_tx,
+        tx: tx,
+        txi: txi,
+        tx_hash: tx_hash,
+        block_index: block_index
+      }) do
+    pk = :aect_contracts.pubkey(:aect_contracts.new(tx))
+    owner_pk = :aect_create_tx.owner_pubkey(tx)
+    :ets.insert(:ct_create_sync_cache, {pk, txi})
+    write_origin(:contract_create_tx, pk, txi, tx_hash)
+    Sync.Contract.create(pk, owner_pk, txi, block_index)
+  end
+
+  def mutate(%__MODULE__{type: :contract_call_tx, tx: tx, txi: txi, block_index: block_index}) do
+    pk = :aect_call_tx.contract_pubkey(tx)
+    Sync.Contract.call(pk, tx, txi, block_index)
+  end
+
+  def mutate(%__MODULE__{
+        type: :channel_create_tx,
+        signed_tx: signed_tx,
+        txi: txi,
+        tx_hash: tx_hash
+      }) do
+    {:ok, pk} = :aesc_utils.channel_pubkey(signed_tx)
+    write_origin(:channel_create_tx, pk, txi, tx_hash)
+  end
+
+  def mutate(%__MODULE__{
+        type: :oracle_register_tx,
+        tx: tx,
+        txi: txi,
+        tx_hash: tx_hash,
+        block_index: block_index
+      }) do
+    pk = :aeo_register_tx.account_pubkey(tx)
+    write_origin(:oracle_register_tx, pk, txi, tx_hash)
+    Sync.Oracle.register(pk, tx, txi, block_index)
+  end
+
+  def mutate(%__MODULE__{type: :oracle_extend_tx, tx: tx, txi: txi, block_index: block_index}) do
+    Sync.Oracle.extend(:aeo_extend_tx.oracle_pubkey(tx), tx, txi, block_index)
+  end
+
+  def mutate(%__MODULE__{type: :oracle_response_tx, tx: tx, txi: txi, block_index: block_index}) do
+    Sync.Oracle.respond(:aeo_response_tx.oracle_pubkey(tx), tx, txi, block_index)
+  end
+
+  def mutate(%__MODULE__{
+        type: :name_claim_tx,
+        tx: tx,
+        txi: txi,
+        tx_hash: tx_hash,
+        block_index: block_index
+      }) do
+    plain_name = String.downcase(:aens_claim_tx.name(tx))
+    {:ok, name_hash} = :aens.get_name_hash(plain_name)
+    write_origin(:name_claim_tx, name_hash, txi, tx_hash)
+    Sync.Name.claim(plain_name, name_hash, tx, txi, block_index)
+  end
+
+  def mutate(%__MODULE__{type: :name_update_tx, tx: tx, txi: txi, block_index: block_index}) do
+    Sync.Name.update(:aens_update_tx.name_hash(tx), tx, txi, block_index)
+  end
+
+  def mutate(%__MODULE__{type: :name_transfer_tx, tx: tx, txi: txi, block_index: block_index}) do
+    Sync.Name.transfer(:aens_transfer_tx.name_hash(tx), tx, txi, block_index)
+  end
+
+  def mutate(%__MODULE__{type: :name_revoke_tx, tx: tx, txi: txi, block_index: block_index}) do
+    Sync.Name.revoke(:aens_revoke_tx.name_hash(tx), tx, txi, block_index)
+  end
+
+  def mutate(_write_links_mutation) do
+    :ok
+  end
+
+  defp write_origin(tx_type, pubkey, txi, tx_hash) do
+    m_origin = Model.origin(index: {tx_type, pubkey, txi}, tx_id: tx_hash)
+    m_rev_origin = Model.rev_origin(index: {txi, tx_type, pubkey})
+    :mnesia.write(Model.Origin, m_origin, :write)
+    :mnesia.write(Model.RevOrigin, m_rev_origin, :write)
+    write_field(tx_type, nil, pubkey, txi)
+  end
+
+  defp write_field(tx_type, pos, pubkey, txi) do
+    m_field = Model.field(index: {tx_type, pos, pubkey, txi})
+    :mnesia.write(Model.Field, m_field, :write)
+    Model.incr_count({tx_type, pos, pubkey})
+  end
+end
+
+defimpl AeMdw.Db.Mutation, for: AeMdw.Db.WriteLinksMutation do
+  def mutate(mutation) do
+    @for.mutate(mutation)
+  end
+end