better error handling when using transaction, #47

Author: zookzook

CHANGELOG.md

@@ -6,6 +6,7 @@
     * refactored the test cases
     * now using mtools for a MongoDB deployment in the travis ci environment
     * travis ci uses only the latest MongoDB version [The failCommand](https://github.com/mongodb/mongo/wiki/The-%22failCommand%22-fail-point)
+    * `Session.commit_transaction` returns now `:ok` or an error `{:error, %Mongo.Error{}}`
 
 * Bugfixes
     * Using `max_staleness_ms` > 0 results in a crash

lib/mongo/error.ex

@@ -108,6 +108,11 @@ defmodule Mongo.Error do
   def should_retry_write(_error, _cmd, _opts) do
     false
   end
+
+  def has_label(%Mongo.Error{error_labels: []}, _label), do: false
+  def has_label(%Mongo.Error{error_labels: labels}, label) do
+    Enum.any?(labels, fn l -> l == label end)
+  end
 end
 
 defmodule Mongo.WriteError do

lib/mongo/events.ex

@@ -67,7 +67,6 @@ defmodule Mongo.Events do
 
   defmodule RetryReadEvent do
     @moduledoc false
-
     defstruct [
       :command,         ## Returns the command.
       :command_name,    ## Returns the command name.

lib/mongo/session.ex

@@ -100,11 +100,14 @@ defmodule Mongo.Session do
   import Mongo.WriteConcern
 
   alias BSON.Timestamp
+  alias Mongo.Error
   alias Mongo.ReadPreference
   alias Mongo.Session
   alias Mongo.Session.ServerSession
   alias Mongo.Topology
 
+  @retry_timeout_seconds 120
+
   @type t :: pid()
 
   ##
@@ -291,19 +294,21 @@ defmodule Mongo.Session do
     with {:ok, session} <- Session.start_session(topology_pid, :write, opts),
          :ok            <- Session.start_transaction(session) do
 
-      with {:ok, result} <- run_function(fun, Keyword.merge(opts, session: session)) do
-        commit_transaction(session)
+      with {:ok, result} <- run_function(fun, Keyword.merge(opts, session: session)),
+           commit_result <- commit_transaction(session) do
+
         end_session(topology_pid, session)
-        {:ok, result}
+        case commit_result do
+          :ok    -> {:ok, result}
+          error -> error
+        end
       else
         error ->
           abort_transaction(session)
           end_session(topology_pid, session)
           error
       end
-
     end
-
   end
 
   ##
@@ -469,7 +474,11 @@ defmodule Mongo.Session do
     {:next_state, :transaction_committed, :ok}
   end
   def handle_call_event(:commit_transaction, :transaction_in_progress, data) do
-    {:next_state, :transaction_committed, run_commit_command(data)}
+    with :ok <- run_commit_command(data) do
+      {:next_state, :transaction_committed, :ok}
+      else
+      error -> {:keep_state_and_data, error}
+    end
   end
   def handle_call_event(:abort_transaction, :starting_transaction, _data) do
     {:next_state, :transaction_aborted, :ok}
@@ -514,21 +523,42 @@ defmodule Mongo.Session do
   ##
   # Run the commit transaction command.
   #
-  defp run_commit_command(%Session{conn: conn, recovery_token: recovery_token, server_session: %ServerSession{session_id: id, txn_num: txn_num}, opts: opts}) do
+  defp run_commit_command(session) do
+    run_commit_command(session, DateTime.utc_now(), :first)
+  end
+
+  defp run_commit_command(%Session{conn: conn,
+    recovery_token: recovery_token,
+    server_session: %ServerSession{session_id: id, txn_num: txn_num},
+    opts: opts} = session, time, n) do
+
+    ##
+    # Drivers should apply a majority write concern when retrying commitTransaction to guard against a transaction being applied twice.
+    write_concern = case n do
+      :first -> write_concern(opts)
+      _      -> Map.put(write_concern(opts) || %{}, :w, :majority)
+    end
 
     cmd = [
             commitTransaction: 1,
             lsid: %{id: id},
             txnNumber: %BSON.LongNumber{value: txn_num},
             autocommit: false,
-            writeConcern: write_concern(opts),
+            writeConcern: write_concern, ## todo: w:majority
             maxTimeMS: max_time_ms(opts),
            recoveryToken: recovery_token
           ] |> filter_nils()
 
-    _doc = Mongo.exec_command(conn, cmd, database: "admin")
-
-    :ok
+   with {:ok, _doc} <- Mongo.exec_command(conn, cmd, database: "admin") do
+     :ok
+   else
+      {:error, error} ->
+        try_again = Error.has_label(error, "UnknownTransactionCommitResult") && DateTime.diff(DateTime.utc_now(), time, :second) < @retry_timeout_seconds
+        case try_again do
+          true  -> run_commit_command(session, time, :retry)
+          false -> {:error, error}
+        end
+    end
   end
 
   defp max_time_ms(opts) do

test/mongo/transaction_retries_test.exs

@@ -0,0 +1,106 @@
+defmodule Mongo.TransactionRetriesTest do
+
+  use CollectionCase
+
+  alias Mongo.Session
+
+  test "transaction returns an error", %{pid: top} do
+
+    coll = unique_collection()
+
+    :ok = Mongo.create(top, coll)
+
+    {:ok, session} = Session.start_session(top, :write, [])
+    assert :ok = Session.start_transaction(session)
+
+    assert {:ok, _} = Mongo.insert_one(top, coll, %{name: "Greta"}, session: session)
+
+    cmd = [
+      configureFailPoint: "failCommand",
+      mode: [times: 1],
+      data: [errorCode: 3, failCommands: ["commitTransaction"]]
+    ]
+
+    {:ok, _doc} = Mongo.admin_command(top, cmd)
+
+    {:error, %Mongo.Error{}} = Session.commit_transaction(session)
+
+    assert :ok == Session.abort_transaction(session)
+    assert :ok == Session.end_session(top, session)
+  end
+
+  @tag :mongo_4_3
+  test "transaction retry", %{pid: top, catcher: catcher} do
+
+    coll = unique_collection()
+
+    :ok = Mongo.create(top, coll)
+
+    {:ok, session} = Session.start_session(top, :write, [])
+    assert :ok = Session.start_transaction(session)
+
+    assert {:ok, _} = Mongo.insert_one(top, coll, %{name: "Greta"}, session: session)
+
+    cmd = [
+      configureFailPoint: "failCommand",
+      mode: [times: 3],
+      data: [errorCode: 6, failCommands: ["commitTransaction"], errorLabels: ["UnknownTransactionCommitResult"]]
+    ]
+
+    {:ok, _doc} = Mongo.admin_command(top, cmd)
+    assert :ok == Session.commit_transaction(session)
+
+    assert :ok == Session.end_session(top, session)
+
+    assert [:commitTransaction, :commitTransaction, :commitTransaction] = EventCatcher.failed_events(catcher) |> Enum.map(fn event -> event.command_name end)
+    assert [:commitTransaction, :configureFailPoint, :insert, :create] = EventCatcher.succeeded_events(catcher) |> Enum.map(fn event -> event.command_name end)
+
+  end
+
+  test "with_transaction, return an error", %{pid: top} do
+
+    coll = unique_collection()
+
+    :ok = Mongo.create(top, coll)
+
+    assert {:error, %Mongo.Error{}} = Session.with_transaction(top, fn opts ->
+      {:ok, _} = Mongo.insert_one(top, coll, %{name: "Greta"}, opts)
+      {:ok, _} = Mongo.insert_one(top, coll, %{name: "Waldo"}, opts)
+      {:ok, _} = Mongo.insert_one(top, coll, %{name: "Tom"}, opts)
+
+      cmd = [
+        configureFailPoint: "failCommand",
+        mode: [times: 1],
+        data: [errorCode: 3, failCommands: ["commitTransaction"]]
+      ]
+
+      {:ok, _doc} = Mongo.admin_command(top, cmd)
+
+      {:ok, []}
+    end)
+  end
+
+  test "with_transaction, retry commit", %{pid: top} do
+
+    coll = unique_collection()
+
+    :ok = Mongo.create(top, coll)
+
+    assert {:ok, []} = Session.with_transaction(top, fn opts ->
+             {:ok, _} = Mongo.insert_one(top, coll, %{name: "Greta"}, opts)
+             {:ok, _} = Mongo.insert_one(top, coll, %{name: "Waldo"}, opts)
+             {:ok, _} = Mongo.insert_one(top, coll, %{name: "Tom"}, opts)
+
+             cmd = [
+               configureFailPoint: "failCommand",
+               mode: [times: 3],
+               data: [errorCode: 6, failCommands: ["commitTransaction"], errorLabels: ["UnknownTransactionCommitResult"]]
+             ]
+
+             {:ok, _doc} = Mongo.admin_command(top, cmd)
+
+             {:ok, []}
+           end)
+  end
+
+end