fix: improved the reconnection workflow

lib/mongo/mongo_db_connection.ex

@@ -37,6 +37,7 @@ defmodule Mongo.MongoDBConnection do
       wire_version: 0,
       auth_mechanism: opts[:auth_mechanism] || nil,
       connection_type: Keyword.fetch!(opts, :connection_type),
+      server_pid: Keyword.get(opts, :server_pid),
       topology_pid: Keyword.fetch!(opts, :topology_pid),
       stable_api: Keyword.get(opts, :stable_api),
       use_op_msg: Keyword.get(opts, :stable_api) != nil,
@@ -48,8 +49,24 @@ defmodule Mongo.MongoDBConnection do
   end
 
   @impl true
-  def disconnect(_error, %{connection: {mod, socket}, connection_type: type, topology_pid: pid, host: host}) do
-    GenServer.cast(pid, {:disconnect, type, host})
+  ## the stream monitor disconnects, we change the mode of the parent monitor
+  def disconnect(_error, %{connection: {mod, socket}, connection_type: :stream_monitor, parent_pid: parent_pid}) do
+    ## Logger.debug("MongoDB-Connection: disconnected stream monitor: #{inspect(error)}")
+    GenServer.cast(parent_pid, :stop_streaming_mode)
+    mod.close(socket)
+    :ok
+  end
+
+  def disconnect(_error, %{connection: {mod, socket}, connection_type: :monitor, topology_pid: topology_pid, host: host, server_pid: server_pid}) do
+    ## Logger.debug("MongoDB-Connection: disconnected: #{inspect(error)}, #{inspect(server_pid)}, #{inspect(host)}, cast disconnect :monitor")
+    GenServer.cast(server_pid, :stop_streaming_mode)
+    GenServer.cast(topology_pid, {:disconnect, :monitor, host, server_pid})
+    mod.close(socket)
+    :ok
+  end
+
+  def disconnect(_error, %{connection: {mod, socket}}) do
+    ## Logger.debug("MongoDB-Connection: disconnected: #{inspect error}, #{inspect type}, #{inspect host} #{inspect server_pid}")
     mod.close(socket)
     :ok
   end

lib/mongo/monitor.ex

@@ -47,6 +47,10 @@ defmodule Mongo.Monitor do
     GenServer.cast(pid, :update)
   end
 
+  def stop_streaming_mode(pid) do
+    GenServer.cast(pid, :stop_streaming_mode)
+  end
+
   def set_heartbeat_frequency_ms(pid, heartbeat_frequency_ms) do
     GenServer.cast(pid, {:update, heartbeat_frequency_ms})
   end
@@ -59,7 +63,7 @@ defmodule Mongo.Monitor do
   Initialize the monitor process
   """
   def init([address, topology_pid, heartbeat_frequency_ms, connection_opts]) do
-    ## debug info("Starting monitor process with pid #{inspect self()}, #{inspect address}")
+    ## Logger.info("Starting monitor process with pid #{inspect(self())}, #{inspect(address)}")
 
     # monitors don't authenticate and use the "admin" database
     opts =
@@ -73,6 +77,7 @@ defmodule Mongo.Monitor do
       |> Keyword.put(:topology_pid, topology_pid)
       |> Keyword.put(:pool_size, 1)
       |> Keyword.put(:idle_interval, 5_000)
+      |> Keyword.put(:server_pid, self())
 
     with {:ok, pid} <- DBConnection.start_link(Mongo.MongoDBConnection, opts) do
       {:ok,
@@ -97,18 +102,17 @@ defmodule Mongo.Monitor do
   end
 
   @doc """
-  In case of terminating we stop the our linked processes as well:
+  In case of terminating we stop our linked processes as well:
   * connection
   * streaming process
   """
   def terminate(reason, %{connection_pid: connection_pid, streaming_pid: nil}) do
-    ## debug info("Terminating monitor for reason #{inspect reason}")
+    ## Logger.debug("Terminating monitor #{inspect(self())} for reason #{inspect(reason)}")
     GenServer.stop(connection_pid, reason)
   end
 
   def terminate(reason, %{connection_pid: connection_pid, streaming_pid: streaming_pid}) do
-    ## debug info("Terminating monitor for reason #{inspect reason}, #{inspect self()}, #{inspect streaming_pid}")
-
+    ## Logger.debug("Terminating monitor #{inspect(self())} for reason #{inspect(reason)}, #{inspect(streaming_pid)}")
     GenServer.stop(connection_pid, reason)
     GenServer.stop(streaming_pid, reason)
   end
@@ -117,6 +121,7 @@ defmodule Mongo.Monitor do
   Report the connection event, so the topology process can now create the connection pool.
   """
   def connected(_connection, me, topology_pid) do
+    ## Logger.info("Monitor #{inspect(me)} connected to server! ")
     Topology.monitor_connected(topology_pid, me)
     GenServer.cast(me, :update)
   end
@@ -125,6 +130,15 @@ defmodule Mongo.Monitor do
     {:reply, Map.put(state, :pid, self()), state}
   end
 
+  def handle_cast(:stop_streaming_mode, %{streaming_pid: streaming_pid} = state) when streaming_pid != nil do
+    spawn(fn -> GenServer.stop(streaming_pid) end)
+    {:noreply, %{state | mode: :polling_mode, streaming_pid: nil}}
+  end
+
+  def handle_cast(:stop_streaming_mode, state) do
+    {:noreply, %{state | mode: :polling_mode}}
+  end
+
   ##
   # Update the server description or the rrt value
   ##
@@ -207,11 +221,11 @@ defmodule Mongo.Monitor do
   # Starts the streaming mode
   ##
   defp start_streaming_mode(%{address: address, topology_pid: topology_pid, opts: opts} = state, _server_description) do
-    args = [topology_pid, address, opts]
+    args = [self(), topology_pid, address, opts]
 
     case StreamingHelloMonitor.start_link(args) do
       {:ok, pid} ->
-        ## debug info("Starting streaming mode: #{inspect self()}")
+        ## Logger.debug("Starting streaming mode: #{inspect(pid)}")
         %{state | mode: :streaming_mode, streaming_pid: pid, heartbeat_frequency_ms: 10_000}
 
       error ->

lib/mongo/password_safe.ex

@@ -10,7 +10,7 @@ defmodule Mongo.PasswordSafe do
 
   use GenServer
 
-  def new() do
+  def start_link() do
     GenServer.start_link(@me, [])
   end
 

lib/mongo/repo.ex

@@ -457,7 +457,7 @@ defmodule Mongo.Repo do
   @callback update(doc :: Mongo.Collection.t(), opts :: Keyword.t()) :: {:ok, Mongo.Collection.t()} | {:error, any()}
 
   @doc """
-  Same as `c:update/1` but raises an error.
+  Same as `c:update/2` but raises an error.
   """
   @callback update!(doc :: Mongo.Collection.t(), opts :: Keyword.t()) :: Mongo.Collection.t()
 
@@ -471,7 +471,7 @@ defmodule Mongo.Repo do
   @callback insert_or_update(doc :: Mongo.Collection.t(), opts :: Keyword.t()) :: {:ok, Mongo.Collection.t()} | {:error, any()}
 
   @doc """
-  Same as `c:insert_or_update/1` but raises an error.
+  Same as `c:insert_or_update/2` but raises an error.
   """
   @callback insert_or_update!(doc :: Mongo.Collection.t(), opts :: Keyword.t()) :: Mongo.Collection.t()
 

lib/mongo/streaming_hello_monitor.ex

@@ -30,14 +30,16 @@ defmodule Mongo.StreamingHelloMonitor do
   @doc """
   Initialize the monitor process
   """
-  def init([topology_pid, address, opts]) do
+  def init([monitor_pid, topology_pid, address, opts]) do
     heartbeat_frequency_ms = 10_000
 
     opts =
       opts
       |> Keyword.drop([:after_connect])
       |> Keyword.put(:after_connect, {StreamingHelloMonitor, :connected, [self()]})
       |> Keyword.put(:connection_type, :stream_monitor)
+      |> Keyword.put(:server_pid, self())
+      |> Keyword.put(:monitor_pid, monitor_pid)
 
     ## debug info("Starting stream hello monitor with options #{inspect(opts, pretty: true)}")
 
@@ -65,7 +67,7 @@ defmodule Mongo.StreamingHelloMonitor do
   In this case we stop the DBConnection.
   """
   def terminate(reason, %{connection_pid: connection_pid}) do
-    ## debug info("Terminating streaming hello monitor for reason #{inspect reason}")
+    ## Logger.debug("Terminating streaming hello monitor for reason #{inspect(reason)}")
     GenServer.stop(connection_pid, reason)
   end
 
@@ -84,7 +86,7 @@ defmodule Mongo.StreamingHelloMonitor do
   end
 
   def handle_info({:EXIT, _pid, reason}, state) do
-    ## debug Logger.warn("Stopped with reason #{inspect reason}")
+    Logger.warning("Stopped with reason #{inspect(reason)}")
     {:stop, reason, state}
   end
 

lib/mongo/topology.ex

@@ -79,7 +79,6 @@ defmodule Mongo.Topology do
     GenServer.call(pid, :get_state)
   end
 
-  # 97
   def select_server(pid, type, opts \\ []) do
     timeout = Keyword.get(opts, :checkout_timeout, @default_checkout_timeout)
     GenServer.call(pid, {:select_server, type, opts}, timeout)
@@ -109,7 +108,7 @@ defmodule Mongo.Topology do
   end
 
   def stop(pid) do
-    GenServer.stop(pid)
+    GenServer.stop(pid, :stop)
   end
 
   ## GenServer Callbacks
@@ -163,14 +162,7 @@ defmodule Mongo.Topology do
     end
   end
 
-  def terminate(_reason, state) do
-    case state.opts[:pw_safe] do
-      nil -> nil
-      pid -> GenServer.stop(pid)
-    end
-
-    Enum.each(state.connection_pools, fn {_address, pid} -> GenServer.stop(pid) end)
-    Enum.each(state.monitors, fn {_address, pid} -> GenServer.stop(pid) end)
+  def terminate(_reason, _state) do
     Mongo.Events.notify(%TopologyClosedEvent{topology_pid: self()})
   end
 
@@ -204,18 +196,25 @@ defmodule Mongo.Topology do
   ##
   # In case of :monitor or :stream_monitor we mark the server description of the address as unknown
   ##
-  def handle_cast({:disconnect, kind, address}, state) when kind in [:monitor, :stream_monitor] do
-    server_description = ServerDescription.parse_hello_response(address, "#{inspect(kind)} disconnected")
+  def handle_cast({:disconnect, :monitor, address, pid}, state) do
+    server_description = ServerDescription.parse_hello_response(address, "monitor disconnected")
+    ## Logger.debug("Disconnect monitor with #{inspect(pid)}")
 
     new_state =
       address
-      |> remove_address(state)
+      |> close_connection_pool(pid, state)
       |> maybe_reinit()
 
     handle_cast({:server_description, server_description}, new_state)
   end
 
-  def handle_cast({:disconnect, _kind, _host}, state) do
+  def handle_cast({:disconnect, :stream_monitor, _host, _pid}, state) do
+    ## IO.inspect("ignored: kind stream_monitor with #{inspect pid}")
+    {:noreply, state}
+  end
+
+  def handle_cast({:disconnect, _kind, _host, _pid}, state) do
+    ## IO.inspect("ignored: kind #{inspect kind}")
     {:noreply, state}
   end
 
@@ -233,6 +232,7 @@ defmodule Mongo.Topology do
 
         {host, ^monitor_pid} ->
           arbiters = fetch_arbiters(state)
+          Mongo.Events.notify(%ServerOpeningEvent{address: host, topology_pid: self()})
 
           if host in arbiters do
             state
@@ -243,8 +243,9 @@ defmodule Mongo.Topology do
               |> Keyword.put(:topology_pid, self())
               |> connect_opts_from_address(host)
 
+            ## Logger.debug("Starting connection pool for #{inspect(host)}")
             {:ok, pool} = DBConnection.start_link(Mongo.MongoDBConnection, conn_opts)
-            connection_pools = Map.put(state.connection_pools, host, pool)
+            connection_pools = replace_pool(state.connection_pools, host, pool)
 
             Process.send_after(self(), {:new_connection, state.waiting_pids}, 10)
 
@@ -279,6 +280,49 @@ defmodule Mongo.Topology do
     {:noreply, state}
   end
 
+  ## remove the address only if the pid is the same
+  defp close_connection_pool(address, pid, state) do
+    ## Logger.debug("Closing connection pool by pid: #{inspect(state.monitors[address] == pid)}, #{inspect(pid)}, #{inspect(state.monitors[address])}")
+
+    case state.monitors[address] == pid do
+      true ->
+        Mongo.Events.notify(%ServerClosedEvent{address: address, topology_pid: self()})
+        ## stopping the connection pool
+        case state.connection_pools[address] do
+          nil ->
+            :ok
+
+          pid ->
+            if Process.alive?(pid) do
+              ## Logger.debug("Stopping the connection pool #{inspect(pid)} für #{inspect(address)}")
+              GenServer.stop(pid)
+            end
+        end
+
+        %{state | connection_pools: Map.delete(state.connection_pools, address)}
+
+      false ->
+        state
+    end
+  end
+
+  ## replaces a pool for the host address
+  defp replace_pool(connection_pools, host, pool) do
+    ## if we found an existing pool, we will stop it first
+    case Map.get(connection_pools, host) do
+      nil ->
+        :noop
+
+      pid ->
+        if Process.alive?(pid) do
+          ## Logger.debug("Stopping the connection pool #{inspect(pid)}")
+          GenServer.stop(pid)
+        end
+    end
+
+    Map.put(connection_pools, host, pool)
+  end
+
   ##
   # Update server description: in case of logical session the function creates a session pool for the `deployment`.
   #
@@ -510,9 +554,6 @@ defmodule Mongo.Topology do
       Enum.reduce(added, state, fn address, state ->
         server_description = state.topology.servers[address]
         connopts = connect_opts_from_address(state.opts, address)
-
-        Mongo.Events.notify(%ServerOpeningEvent{address: address, topology_pid: self()})
-
         args = [server_description.address, self(), heartbeat_frequency_ms, Keyword.put(connopts, :pool, DBConnection.ConnectionPool)]
         {:ok, pid} = Monitor.start_link(args)
 
@@ -549,16 +590,23 @@ defmodule Mongo.Topology do
   end
 
   defp remove_address(address, state) do
-    Mongo.Events.notify(%ServerClosedEvent{address: address, topology_pid: self()})
-
     case state.monitors[address] do
-      nil -> :ok
-      pid -> GenServer.stop(pid)
+      nil ->
+        :ok
+
+      pid ->
+        Mongo.Events.notify(%ServerClosedEvent{address: address, topology_pid: self()})
+        ## Logger.debug("Stopping: #{inspect(pid)} for #{inspect(address)}")
+        GenServer.stop(pid)
     end
 
     case state.connection_pools[address] do
-      nil -> :ok
-      pid -> GenServer.stop(pid)
+      nil ->
+        :ok
+
+      pid ->
+        ## Logger.debug("Connection pool: #{inspect(address)}")
+        GenServer.stop(pid)
     end
 
     %{state | monitors: Map.delete(state.monitors, address), connection_pools: Map.delete(state.connection_pools, address)}

lib/mongo/url_parser.ex

@@ -168,7 +168,7 @@ defmodule Mongo.UrlParser do
 
       value ->
         ## start GenServer and put id
-        with {:ok, pid} <- Mongo.PasswordSafe.new(),
+        with {:ok, pid} <- Mongo.PasswordSafe.start_link(),
              :ok <- Mongo.PasswordSafe.set_password(pid, value) do
           opts
           |> Keyword.put(:password, "*****")

mix.exs

@@ -37,7 +37,7 @@ defmodule Mongodb.Mixfile do
       {:patch, "~> 0.12.0", only: [:dev, :test]},
       {:jason, "~> 1.3", only: [:dev, :test]},
       {:credo, "~> 1.7.0", only: [:dev, :test], runtime: false},
-      {:ex_doc, ">= 0.0.0", only: :dev, runtime: false}
+      {:ex_doc, "== 0.24.1", only: :dev, runtime: false}
     ]
   end
 

mix.lock

@@ -4,7 +4,7 @@
   "db_connection": {:hex, :db_connection, "2.6.0", "77d835c472b5b67fc4f29556dee74bf511bbafecdcaf98c27d27fa5918152086", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "c2f992d15725e721ec7fbc1189d4ecdb8afef76648c746a8e1cad35e3b8a35f3"},
   "decimal": {:hex, :decimal, "2.1.1", "5611dca5d4b2c3dd497dec8f68751f1f1a54755e8ed2a966c2633cf885973ad6", [:mix], [], "hexpm", "53cfe5f497ed0e7771ae1a475575603d77425099ba5faef9394932b35020ffcc"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.39", "424642f8335b05bb9eb611aa1564c148a8ee35c9c8a8bba6e129d51a3e3c6769", [:mix], [], "hexpm", "06553a88d1f1846da9ef066b87b57c6f605552cfbe40d20bd8d59cc6bde41944"},
-  "ex_doc": {:hex, :ex_doc, "0.31.1", "8a2355ac42b1cc7b2379da9e40243f2670143721dd50748bf6c3b1184dae2089", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.1", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "3178c3a407c557d8343479e1ff117a96fd31bafe52a039079593fb0524ef61b0"},
+  "ex_doc": {:hex, :ex_doc, "0.24.1", "15673de99154f93ca7f05900e4e4155ced1ee0cd34e0caeee567900a616871a4", [:mix], [{:earmark_parser, "~> 1.4.0", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "07972f17bdf7dc7b5bd76ec97b556b26178ed3f056e7ec9288eb7cea7f91cce2"},
   "file_system": {:hex, :file_system, "1.0.0", "b689cc7dcee665f774de94b5a832e578bd7963c8e637ef940cd44327db7de2cd", [:mix], [], "hexpm", "6752092d66aec5a10e662aefeed8ddb9531d79db0bc145bb8c40325ca1d8536d"},
   "jason": {:hex, :jason, "1.4.1", "af1504e35f629ddcdd6addb3513c3853991f694921b1b9368b0bd32beb9f1b63", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "fbb01ecdfd565b56261302f7e1fcc27c4fb8f32d56eab74db621fc154604a7a1"},
   "makeup": {:hex, :makeup, "1.1.1", "fa0bc768698053b2b3869fa8a62616501ff9d11a562f3ce39580d60860c3a55e", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "5dc62fbdd0de44de194898b6710692490be74baa02d9d108bc29f007783b0b48"},

test/mongo/password_safe_test.exs

@@ -7,7 +7,7 @@ defmodule Mongo.PasswordSafeTest do
 
   test "encrypted password" do
     pw = "my-secret-password"
-    {:ok, pid} = PasswordSafe.new()
+    {:ok, pid} = PasswordSafe.start_link()
     PasswordSafe.set_password(pid, pw)
     %{key: _key, pw: enc_pw} = :sys.get_state(pid)
     assert enc_pw != pw