refactor: commit only through mutations (#534)

Author: jyeshe

lib/ae_mdw/blocks.ex

@@ -17,7 +17,7 @@ defmodule AeMdw.Blocks do
   @type height() :: non_neg_integer()
   @type mbi() :: non_neg_integer()
   @type time() :: non_neg_integer()
-  @type block_index() :: {height(), mbi()}
+  @type block_index() :: {height(), mbi() | -1}
   @type txi_pos() :: non_neg_integer() | -1
   @type block_index_txi_pos() :: {height(), txi_pos()}
   @type key_header() :: term()

lib/ae_mdw/database.ex

@@ -13,6 +13,7 @@ defmodule AeMdw.Database do
 
   alias AeMdw.Db.Model
   alias AeMdw.Db.Mutation
+  alias AeMdw.Db.TxnMutation
   alias AeMdw.Db.RocksDb
   alias AeMdw.Db.RocksDbCF
 
@@ -22,6 +23,7 @@ defmodule AeMdw.Database do
   @type direction() :: :forward | :backward
   @type cursor() :: key() | nil
   @type limit() :: pos_integer()
+  @type transaction() :: RocksDb.transaction()
 
   @end_token :"$end_of_table"
 
@@ -209,18 +211,29 @@ defmodule AeMdw.Database do
     :mnesia.read(table, key, lock)
   end
 
-  @spec write(table(), record()) :: :ok
-  def write(tab, record) when use_rocksdb?(tab) do
-    RocksDbCF.put(tab, record)
+  @spec write(transaction(), table(), record()) :: :ok
+  def write(txn, tab, record) when use_rocksdb?(tab) do
+    RocksDbCF.put(txn, tab, record)
   end
 
+  @spec write(table(), record()) :: :ok
   def write(table, record) do
     :mnesia.write(table, record, :write)
   end
 
-  @spec commit() :: :ok
-  def commit do
-    :ok = RocksDb.commit()
+  @doc """
+  Creates a transaction and commits the changes of a mutation list.
+  """
+  @spec commit([TxnMutation.t()]) :: :ok
+  def commit(txn_mutations) do
+    {:ok, txn} = RocksDb.transaction_new()
+
+    txn_mutations
+    |> List.flatten()
+    |> Enum.reject(&is_nil/1)
+    |> Enum.each(&TxnMutation.execute(&1, txn))
+
+    :ok = RocksDb.transaction_commit(txn)
   end
 
   @spec transaction([Mutation.t()]) :: :ok

lib/ae_mdw/db/mutations/key_blocks_mutation.ex

@@ -20,18 +20,18 @@ defmodule AeMdw.Db.KeyBlocksMutation do
     %__MODULE__{key_block: m_block, next_txi: next_txi}
   end
 
-  @spec mutate(t()) :: :ok
-  def mutate(%__MODULE__{key_block: m_block, next_txi: next_txi}) do
+  @spec execute(t(), Database.transaction()) :: :ok
+  def execute(%__MODULE__{key_block: m_block, next_txi: next_txi}, txn) do
     {height, -1} = Model.block(m_block, :index)
     [next_kb] = Database.read(Model.Block, {height + 1, -1})
 
-    Database.write(Model.Block, m_block)
-    Database.write(Model.Block, Model.block(next_kb, tx_index: next_txi))
+    Database.write(txn, Model.Block, m_block)
+    Database.write(txn, Model.Block, Model.block(next_kb, tx_index: next_txi))
   end
 end
 
-defimpl AeMdw.Db.Mutation, for: AeMdw.Db.KeyBlocksMutation do
-  def mutate(mutation) do
-    @for.mutate(mutation)
+defimpl AeMdw.Db.TxnMutation, for: AeMdw.Db.KeyBlocksMutation do
+  def execute(mutation, txn) do
+    @for.execute(mutation, txn)
   end
 end

lib/ae_mdw/db/mutations/mnesia_write_mutation.ex renamed to lib/ae_mdw/db/mutations/write_mutation.ex

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