fix timeout value if using the streaming_hello_monitor.ex, clarify usage

Author: zookzook

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Bugfix
   * fix a crash in the streaming hello monitor, if the server sends more than one response at once 
   * add support for the new hello handshake
+  * refactor :timeout option (thanks to JD-Robertson for reporting)
 
 ## 0.9.1 (2022-05-27)
 * Bugfix

README.md

@@ -488,6 +488,26 @@ you'll want to add this cipher to your `ssl_opts`:
 )
 ```
 
+## Timeout
+
+The `:timeout` option is only used to specify the timeout for receiving data on the tcp layer. The default value is 15 seconds. 
+The connection pool defines separate timeout values. 
+You can use the `:timeout` as a global option to override the default value:
+
+```elixir
+# Starts an pooled connection
+{:ok, conn} = Mongo.start_link(url: "mongodb://localhost:27017/db-name", timeout: 60_000)
+```
+
+Each single connection uses `60_000` (60 seconds) as the timeout value instead of `15_000`. But you can override the default value by
+using the `:timeout` option, when running a single command:
+
+```elixr
+Mongo.find(conn, "dogs", %{}, timeout: 120_000)
+```
+
+Now the driver will use 120 seconds for the timeout.
+
 ## Change Streams
 
 Change streams are available in replica set and sharded cluster deployments

lib/mongo.ex

@@ -67,7 +67,6 @@ defmodule Mongo do
   alias Mongo.Events.CommandFailedEvent
   alias Mongo.Error
 
-  @timeout 15_000
   @retry_timeout_seconds 120
 
   @type conn :: DbConnection.Conn
@@ -1504,7 +1503,7 @@ defmodule Mongo do
           {: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),
-         {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:command, new_cmd}}, [], defaults(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
       {:ok, doc}
@@ -1518,25 +1517,25 @@ defmodule Mongo do
   @spec exec_command(GenServer.server(), BSON.document(), Keyword.t()) ::
           {:ok, {any(), BSON.document()} | nil} | {:error, Mongo.Error.t()}
   def exec_command(conn, cmd, opts) do
-    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:command, cmd}}, [], defaults(opts)) do
+    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:command, cmd}}, [], opts) do
       check_for_error(response, cmd, opts)
     end
   end
 
   def exec_hello(conn, opts) do
-    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:exec_hello, []}}, [], defaults(opts)) do
+    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:exec_hello, []}}, [], opts) do
       check_for_error(response, [hello: 1], opts)
     end
   end
 
   def exec_hello(conn, cmd, opts) do
-    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:exec_hello, cmd}}, [], defaults(opts)) do
+    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:exec_hello, cmd}}, [], opts) do
       check_for_error(response, cmd, opts)
     end
   end
 
   def exec_more_to_come(conn, opts) do
-    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: :more_to_come}, [], defaults(opts)) do
+    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: :more_to_come}, [], opts) do
       check_for_error(response, [:more_to_come], opts)
     end
   end
@@ -1747,10 +1746,6 @@ defmodule Mongo do
   defp assert_many_docs!(other),
     do: raise(ArgumentError, "expected list of documents, got: #{inspect(other)}")
 
-  defp defaults(opts) do
-    Keyword.put_new(opts, :timeout, @timeout)
-  end
-
   ## support for logging like Ecto
   defp do_log([:more_to_come], _duration, _opts) do
     :ok

lib/mongo/mongo_db_connection.ex

@@ -14,7 +14,7 @@ defmodule Mongo.MongoDBConnection do
   alias Mongo.Events.MoreToComeEvent
   alias Mongo.StableVersion
 
-  @timeout 5_000
+  @timeout 15_000
   @find_one_flags ~w(slave_ok exhaust partial)a
   @write_concern ~w(w j wtimeout)a
   @insecure_cmds [:authenticate, :saslStart, :saslContinue, :getnonce, :createUser, :updateUser, :copydbgetnonce, :copydbsaslstart, :copydb, :isMaster, :ismaster, :hello]

lib/mongo/streaming_hello_monitor.ex

@@ -151,10 +151,15 @@ defmodule Mongo.StreamingHelloMonitor do
   end
 
   defp hello_command(conn_pid, %{"counter" => counter, "processId" => process_id}, opts) do
-    opts = Keyword.merge(opts, flags: [:exhaust_allowed])
+    maxAwaitTimeMS = Keyword.get(opts, :max_await_time_ms, 10_000)
+
+    opts =
+      opts
+      |> Keyword.merge(flags: [:exhaust_allowed])
+      |> Keyword.merge(timeout: maxAwaitTimeMS * 2)
 
     cmd = [
-      maxAwaitTimeMS: Keyword.get(opts, :max_await_time_ms, 10_000),
+      maxAwaitTimeMS: maxAwaitTimeMS,
       topologyVersion: %{
         counter: %BSON.LongNumber{value: counter},
         processId: process_id

lib/mongo_db_connection/utils.ex

@@ -126,7 +126,6 @@ defmodule Mongo.MongoDBConnection.Utils do
         {:ok, id, reply}
 
       :error ->
-        ## todo
         case mod.recv(socket, 0, state.timeout) do
           {:ok, tail} -> recv_data(header, [data | tail], state)
           {:error, reason} -> recv_error(reason, state)