fix: applies the global timeout value to each query (#215)

Author: zookzook

lib/mongo.ex

@@ -1502,7 +1502,7 @@ defmodule Mongo do
  @spec exec_command_session(GenServer.server(), BSON.document(), Keyword.t()) ::
  {:ok, BSON.document() | nil} | {:error, Mongo.Error.t()}
  def exec_command_session(session, cmd, opts) do
- with {:ok, conn, new_cmd} <- Session.bind_session(session, cmd),
+ with {:ok, conn, new_cmd, opts} <- Session.bind_session(session, cmd, opts),
  {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:command, new_cmd}}, [], opts),
  :ok <- Session.update_session(session, response, opts),
  {:ok, {_flags, doc}} <- check_for_error(response, cmd, opts) do

lib/mongo/monitor.ex

@@ -186,18 +186,8 @@ defmodule Mongo.Monitor do
  ##
  # Get a new server description from the server and send it to the Topology process.
  ##
- defp update_server_description(%{topology_pid: topology_pid, address: address, mode: :streaming_mode} = state) do
- case get_server_description(state) do
- %{round_trip_time: round_trip_time} ->
- ## debug info("Updating round_trip_time: #{inspect round_trip_time}")
- Topology.update_rrt(topology_pid, address, round_trip_time)
-
- %{state | round_trip_time: round_trip_time}
-
- error ->
- warning("Unable to round trip time because of #{inspect(error)}")
- state
- end
+ defp update_server_description(%{mode: :streaming_mode} = state) do
+ state
  end
 
  ##

lib/mongo/session.ex

@@ -186,14 +186,15 @@ defmodule Mongo.Session do
 
  @doc """
  Merge the session / transaction data into the cmd. There is no need to call this function directly. It is called automatically.
+ The global session timeout is merged to the options as well.
  """
- @spec bind_session(Session.t(), BSON.document()) :: {:ok, pid, BSON.document()} | {:error, term()}
- def bind_session(nil, _cmd) do
+ @spec bind_session(Session.t(), BSON.document(), Keyword.t()) :: {:ok, pid, BSON.document(), Keyword.t()} | {:error, term()}
+ def bind_session(nil, _cmd, _opts) do
  {:error, Mongo.Error.exception("No session")}
  end
 
- def bind_session(pid, cmd) do
- call(pid, {:bind_session, cmd})
+ def bind_session(pid, cmd, opts) do
+ call(pid, {:bind_session, cmd, opts})
  end
 
  @doc """
@@ -462,13 +463,16 @@ defmodule Mongo.Session do
  ##
  # bind session: only if wire_version >= 6, MongoDB 3.6.x and no transaction is running: only lsid and the transaction-id is added
  #
- def handle_call_event({:bind_session, cmd}, transaction, %Session{conn: conn, opts: opts, wire_version: wire_version, server_session: %ServerSession{session_id: id, txn_num: txn_num}} = data)
+ def handle_call_event({:bind_session, cmd, client_opts}, transaction, %Session{conn: conn, opts: opts, wire_version: wire_version, server_session: %ServerSession{session_id: id, txn_num: txn_num}} = data)
  when wire_version >= 6 and transaction in [:no_transaction, :transaction_aborted, :transaction_committed] do
  ## only if retryable_writes are enabled!
  options =
  case opts[:retryable_writes] do
- true -> [lsid: %{id: id}, txnNumber: %BSON.LongNumber{value: txn_num}, readConcern: read_concern(data, Keyword.get(cmd, :readConcern))]
- _ -> [lsid: %{id: id}, readConcern: read_concern(data, Keyword.get(cmd, :readConcern))]
+ true ->
+ [lsid: %{id: id}, txnNumber: %BSON.LongNumber{value: txn_num}, readConcern: read_concern(data, Keyword.get(cmd, :readConcern))]
+
+ _ ->
+ [lsid: %{id: id}, readConcern: read_concern(data, Keyword.get(cmd, :readConcern))]
  end
 
  cmd =
@@ -477,11 +481,12 @@ defmodule Mongo.Session do
  |> ReadPreference.add_read_preference(opts)
  |> filter_nils()
 
- {:keep_state_and_data, {:ok, conn, cmd}}
+ client_opts = merge_timeout(client_opts, opts)
+ {:keep_state_and_data, {:ok, conn, cmd, client_opts}}
  end
 
- def handle_call_event({:bind_session, cmd}, :starting_transaction, %Session{conn: conn, server_session: %ServerSession{session_id: id, txn_num: txn_num}, wire_version: wire_version} = data) when wire_version >= 6 do
- result =
+ def handle_call_event({:bind_session, cmd, client_opts}, :starting_transaction, %Session{conn: conn, opts: opts, server_session: %ServerSession{session_id: id, txn_num: txn_num}, wire_version: wire_version} = data) when wire_version >= 6 do
+ cmd =
  Keyword.merge(cmd,
  readConcern: read_concern(data, Keyword.get(cmd, :readConcern)),
  lsid: %{id: id},
@@ -492,10 +497,11 @@ defmodule Mongo.Session do
  |> filter_nils()
  |> Keyword.drop(~w(writeConcern)a)
 
- {:next_state, :transaction_in_progress, {:ok, conn, result}}
+ client_opts = merge_timeout(client_opts, opts)
+ {:next_state, :transaction_in_progress, {:ok, conn, cmd, client_opts}}
  end
 
- def handle_call_event({: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_call_event({:bind_session, cmd, client_opts}, :transaction_in_progress, %Session{conn: conn, opts: opts, 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},
@@ -504,12 +510,13 @@ defmodule Mongo.Session do
  )
  |> Keyword.drop(~w(writeConcern readConcern)a)
 
- {:keep_state_and_data, {:ok, conn, result}}
+ client_opts = merge_timeout(client_opts, opts)
+ {:keep_state_and_data, {:ok, conn, result, client_opts}}
  end
 
  # In case of wire_version < 6 we do nothing
- def handle_call_event({:bind_session, cmd}, _transaction, %Session{conn: conn}) do
- {:keep_state_and_data, {:ok, conn, cmd}}
+ def handle_call_event({:bind_session, cmd, client_opts}, _transaction, %Session{conn: conn}) do
+ {:keep_state_and_data, {:ok, conn, cmd, client_opts}}
  end
 
  def handle_call_event({:commit_transaction, _start_time}, :starting_transaction, _data) do
@@ -710,4 +717,14 @@ defmodule Mongo.Session do
  def in_session(session, _topology_pid, _read_write_type, fun, opts) do
  fun.(session, opts)
  end
+
+ defp merge_timeout(opts, default_ops) do
+ case Keyword.get(default_ops, :timeout) do
+ nil ->
+ opts
+
+ timeout ->
+ Keyword.put_new(opts, :timeout, timeout)
+ end
+ end
 end

lib/mongo/topology.ex

@@ -383,6 +383,8 @@ defmodule Mongo.Topology do
 
  ## found
  {:ok, {address, opts}} ->
+ opts = merge_timeout(opts, state.opts)
+
  with {:ok, connection} <- get_connection(address, state),
  wire_version <- wire_version(address, topology),
  {server_session, new_state} <- checkout_server_session(state),
@@ -593,4 +595,14 @@ defmodule Mongo.Topology do
  Keyword.put_new(opts, :read_preference, read_preference)
  end
  end
+
+ defp merge_timeout(opts, default_ops) do
+ case Keyword.get(default_ops, :timeout) do
+ nil ->
+ opts
+
+ timeout ->
+ Keyword.put_new(opts, :timeout, timeout)
+ end
+ end
 end