better support for errors while using bulk write, added some new tests for transaction, added missing write concern options

Author: zookzook

lib/mongo.ex

@@ -279,20 +279,22 @@ defmodule Mongo do
   @spec find_one_and_update(GenServer.server, collection, BSON.document, BSON.document, Keyword.t) :: result(BSON.document) | {:ok, nil}
   def find_one_and_update(topology_pid, coll, filter, update, opts \\ []) do
     _ = modifier_docs(update, :update)
+
     cmd = [
             findAndModify:            coll,
             query:                    filter,
+            sort:                     opts[:sort],
             update:                   update,
-            bypassDocumentValidation: opts[:bypass_document_validation],
-            maxTimeMS:                opts[:max_time],
-            fields:                   opts[:projection],
             new:                      should_return_new(opts[:return_document]),
-            sort:                     opts[:sort],
+            fields:                   opts[:projection],
             upsert:                   opts[:upsert],
+            bypassDocumentValidation: opts[:bypass_document_validation],
+            writeConcern:             write_concern(opts),
+            maxTimeMS:                opts[:max_time],
             collation:                opts[:collation]
           ] |> filter_nils()
 
-    opts = Keyword.drop(opts, ~w(bypass_document_validation max_time projection return_document sort upsert collation)a)
+    opts = Keyword.drop(opts, ~w(bypass_document_validation max_time projection return_document sort upsert collation w j wtimeout)a)
 
     with {:ok, doc} <- issue_command(topology_pid, cmd, :write, opts) do
         {:ok, doc["value"]}
@@ -342,6 +344,9 @@ defmodule Mongo do
   @spec find_one_and_replace(GenServer.server, collection, BSON.document, BSON.document, Keyword.t) :: result(BSON.document)
   def find_one_and_replace(topology_pid, coll, filter, replacement, opts \\ []) do
     _ = modifier_docs(replacement, :replace)
+
+    write_concern = write_concern(opts)
+
     cmd = [
             findAndModify:            coll,
             query:                    filter,
@@ -352,7 +357,8 @@ defmodule Mongo do
             new:                      should_return_new(opts[:return_document]),
             sort:                     opts[:sort],
             upsert:                   opts[:upsert],
-            collation:                opts[:collation]
+            collation:                opts[:collation],
+            writeConcern:             write_concern
           ] |> filter_nils()
 
     opts = Keyword.drop(opts, ~w(bypass_document_validation max_time projection return_document sort upsert collation)a)
@@ -376,14 +382,18 @@ defmodule Mongo do
   """
   @spec find_one_and_delete(GenServer.server, collection, BSON.document, Keyword.t) :: result(BSON.document)
   def find_one_and_delete(topology_pid, coll, filter, opts \\ []) do
+
+    write_concern = write_concern(opts)
+
     cmd = [
             findAndModify: coll,
             query:         filter,
             remove:        true,
             maxTimeMS:     opts[:max_time],
             fields:        opts[:projection],
             sort:          opts[:sort],
-            collation:     opts[:collation]
+            collation:     opts[:collation],
+            writeConcern:  write_concern
           ] |> filter_nils()
     opts = Keyword.drop(opts, ~w(max_time projection sort collation)a)
 
@@ -437,8 +447,8 @@ defmodule Mongo do
 
     case documents do
       [%{"n" => count}] -> {:ok, count}
-      []                -> {:error, Mongo.Error.exception(message: "nothing returned")}
-      _                 -> :ok # fixes {:error, :too_many_documents_returned}
+      []                -> {:ok, 0}
+      _                 -> :error
     end
   end
 

lib/mongo/bulk_write.ex

@@ -298,6 +298,7 @@ defmodule Mongo.BulkWrite do
     |> Enum.map(fn
       {:ok, %{"n" => n} = doc} -> BulkWriteResult.insert_result(n, ids, doc["writeErrors"] || [])
       {:ok, _other}            -> BulkWriteResult.empty()
+      {:error, reason}         -> BulkWriteResult.error(reason)
     end)
     |> BulkWriteResult.reduce()
   end
@@ -310,6 +311,7 @@ defmodule Mongo.BulkWrite do
                                                                               BulkWriteResult.update_result(n - l, modified, l, filter_upsert_ids(ids), doc["writeErrors"] || [])
       {:ok, %{"n" => matched, "nModified" => modified} = doc}              -> BulkWriteResult.update_result(matched, modified, 0, [], doc["writeErrors"] || [])
       {:ok, _other}                                                        -> BulkWriteResult.empty()
+      {:error, reason}                                                     -> BulkWriteResult.error(reason)
     end)
     |> BulkWriteResult.reduce()
 
@@ -320,6 +322,7 @@ defmodule Mongo.BulkWrite do
     |> Enum.map(fn
       {:ok, %{"n" => n} = doc } -> BulkWriteResult.delete_result(n, doc["writeErrors"] || [])
       {:ok, _other}             -> BulkWriteResult.empty()
+      {:error, reason}          -> BulkWriteResult.error(reason)
     end)
     |> BulkWriteResult.reduce()
 

lib/mongo/results.ex

@@ -108,6 +108,10 @@ defmodule Mongo.BulkWriteResult do
     %BulkWriteResult{deleted_count: count, errors: errors}
   end
 
+  def error(error) do
+    %BulkWriteResult{errors: [error]}
+  end
+
   def empty() do
     %BulkWriteResult{}
   end

lib/mongo/session.ex

@@ -2,15 +2,15 @@ defmodule Mongo.Session do
 
   @moduledoc """
 
-    see https://github.com/mongodb/specifications/blob/master/source/transactions/transactions.rst#committransaction
-
-
-    see https://andrealeopardi.com/posts/connection-managers-with-gen_statem/
+  For gen_statem look here
+  * see https://github.com/mongodb/specifications/blob/master/source/transactions/transactions.rst#committransaction
+  * see https://andrealeopardi.com/posts/connection-managers-with-gen_statem/
   """
 
   @behaviour :gen_statem
 
   import Keywords
+  import Mongo.WriteConcern
 
   alias Mongo.Session.ServerSession
   alias Mongo.Session
@@ -114,6 +114,39 @@ defmodule Mongo.Session do
     end
   end
 
+  def with_transaction(topology_pid, fun, opts \\ []) 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)
+        end_session(topology_pid, session)
+        {:ok, result}
+      else
+        error ->
+          abort_transaction(session)
+          end_session(topology_pid, session)
+          ## todo rerun
+          error
+      end
+
+    end
+
+  end
+
+
+  defp run_function(fun, opts) do
+
+    ## warte max 120ms, ansonsten kill
+    try do
+    rescue
+      reason -> {:error, reason}
+    end
+    fun.(opts)
+
+  end
+
   def connection(pid) do
     :gen_statem.call(pid, {:connection})
   end
@@ -136,43 +169,71 @@ defmodule Mongo.Session do
   end
 
   @impl true
-  def handle_event({:call, from}, {:start_transaction}, state, %Session{server_session: session} = data) when state in [:no_transaction, :transaction_aborted, :transaction_committed] do
+  def handle_event({:call, from},
+        {:start_transaction},
+        transaction,
+        %Session{server_session: session} = data) when transaction in [:no_transaction, :transaction_aborted, :transaction_committed] do
     {:next_state, :starting_transaction, %Session{data | server_session: ServerSession.next_txn_num(session)}, {:reply, from, :ok}}
   end
+
   ##
-  # bind session: only if wire_version >= 6, MongoDB 3.6.x
+  # bind session: only if wire_version >= 6, MongoDB 3.6.x and no transaction is running: only lsid is added
   #
-  def handle_event({:call, from}, {:bind_session, cmd}, :no_transaction, %Session{conn: conn, wire_version: wire_version, server_session: %ServerSession{session_id: id}}) when wire_version >= 6 do
+  def handle_event({:call, from},
+        {:bind_session, cmd},
+        transaction,
+        %Session{conn: conn,
+          wire_version: wire_version,
+          server_session: %ServerSession{session_id: id}}) when wire_version >= 6 and transaction in [:no_transaction, :transaction_aborted, :transaction_committed] do
     {:keep_state_and_data, {:reply, from, {:ok, conn, Keyword.merge(cmd, lsid: %{id: id})}}}
   end
-  def handle_event({:call, from}, {:bind_session, cmd}, :starting_transaction,
-                    %Session{conn: conn,
-                      server_session: %ServerSession{session_id: id, txn_num: txn_num},
-                      wire_version: wire_version, opts: opts} = data) when wire_version >= 6 do
+
+  def handle_event({:call, from},
+        {:bind_session, cmd},
+        :starting_transaction,
+        %Session{conn: conn,
+          server_session: %ServerSession{session_id: id, txn_num: txn_num},
+          wire_version: wire_version,
+          opts: opts} = data) when wire_version >= 6 do
     result = Keyword.merge(cmd,
                            readConcern: Keyword.get(opts, :read_concern),
                            lsid: %{id: id},
                            txnNumber: %BSON.LongNumber{value: txn_num},
                            startTransaction: true,
-                           autocommit: false) |> filter_nils()
+                           autocommit: false) |> filter_nils() |> Keyword.drop(~w(writeConcern)a)
+
     {:next_state, :transaction_in_progress, data, {:reply, from, {:ok, conn, result}}}
   end
-  def handle_event({:call, from}, {:bind_session, cmd}, :transaction_in_progress,
-                    %Session{conn: conn, wire_version: wire_version,
-                      server_session: %ServerSession{session_id: id, txn_num: txn_num}}) when wire_version >= 6 do
+
+  def handle_event({:call, from},
+        {:bind_session, cmd},
+        :transaction_in_progress,
+        %Session{conn: conn, wire_version: wire_version,
+          server_session: %ServerSession{session_id: id, txn_num: txn_num}}) when wire_version >= 6 do
     result = Keyword.merge(cmd,
                            lsid: %{id: id},
                            txnNumber: %BSON.LongNumber{value: txn_num},
-                           autocommit: false)
+                           autocommit: false) |> Keyword.drop(~w(writeConcern readConcern)a)
     {:keep_state_and_data, {:reply, from, {:ok, conn, result}}}
   end
-  def handle_event({:call, from}, {:bind_session, cmd}, transaction, %Session{conn: conn}) when transaction in [:no_transaction, :starting_transaction, :transaction_in_progress] do
+
+  # In case of wire_version < 6 we do nothing
+  def handle_event({:call, from},
+        {:bind_session, cmd},
+        _transaction,
+        %Session{conn: conn}) do
     {:keep_state_and_data, {:reply, from, {:ok, conn, cmd}}}
   end
 
+  def handle_event({:call, from}, {:commit_transaction}, :starting_transaction, data) do
+    {:next_state, :transaction_committed, data, {:reply, from, :ok}}
+  end
   def handle_event({:call, from}, {:commit_transaction}, :transaction_in_progress, data) do
     {:next_state, :transaction_committed, data, {:reply, from, run_commit_command(data)}}
   end
+  def handle_event({:call, from}, {:abort_transaction}, :starting_transaction, data) do
+    {:next_state, :transaction_aborted, data, {:reply, from, :ok}}
+  end
   def handle_event({:call, from}, {:abort_transaction}, :transaction_in_progress, data) do
     {:next_state, :transaction_aborted, data, {:reply, from, run_abort_command(data)}}
   end
@@ -215,7 +276,7 @@ defmodule Mongo.Session do
       lsid: %{id: id},
       txnNumber: %BSON.LongNumber{value: txn_num},
       autocommit: false,
-      writeConcern: Keyword.get(opts, :write_concern),
+      writeConcern: write_concern(opts),
       maxTimeMS: Keyword.get(opts, :max_commit_time_ms)
       ] |> filter_nils()
 
@@ -225,12 +286,13 @@ defmodule Mongo.Session do
   defp run_abort_command(%{conn: conn, server_session: %ServerSession{session_id: id, txn_num: txn_num}, opts: opts}) do
 
     Logger.debug("Running abort transaction")
+
     cmd = [
       abortTransaction: 1,
       lsid: %{id: id},
       txnNumber: %BSON.LongNumber{value: txn_num},
       autocommit: false,
-      writeConcern: Keyword.get(opts, :write_concern),
+      writeConcern: write_concern(opts)
     ] |> filter_nils()
 
     Mongo.exec_command(conn, cmd, database: "admin")