not master wip

Author: p

lib/mongo.ex

@@ -407,7 +407,7 @@ defmodule Mongo do
     opts = Mongo.retryable_reads(opts)
 
     with {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
-         result <- exec_command_session(session, cmd, opts),
+         result <- exec_command_session(topology_pid, session, cmd, opts),
          :ok <- Session.end_implict_session(topology_pid, session) do
       case result do
         {:error, error} ->
@@ -429,7 +429,7 @@ defmodule Mongo do
     opts = Mongo.retryable_writes(opts, acknowledged?(cmd[:writeConcerns]))
 
     with {:ok, session} <- Session.start_implicit_session(topology_pid, :write, opts),
-         result         <- exec_command_session(session, cmd, opts),
+         result         <- exec_command_session(topology_pid, session, cmd, opts),
          :ok            <- Session.end_implict_session(topology_pid, session) do
       result
     else
@@ -723,20 +723,28 @@ defmodule Mongo do
   end
 
   @doc false
-  @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),
+  @spec exec_command_session(GenServer.server, GenServer.server, BSON.document, Keyword.t) :: {:ok, BSON.document | nil} | {:error, Mongo.Error.t}
+  def exec_command_session(topology_pid, session, cmd, opts) do
+    with {:ok, conn, new_cmd, address}      <- Session.bind_session(session, cmd),
          {:ok, _cmd, {doc, event}} <- DBConnection.execute(conn, %Query{action: :command}, [new_cmd], defaults(opts)),
          doc                       <- Session.update_session(session, doc, opts),
-         {:ok, doc}                <- check_for_error(doc, event) do
+         {:ok, doc}                <- check_for_error(doc, event, address) do
       {:ok, doc}
     else
       {:error, error} ->
       ## todo update Topology
+      IO.inspect(error)
+        if error.not_master_or_recovering do
+          IO.inspect("ok")
+          server_description = Mongo.ServerDescription.from_is_master_error(error.address, error)
+          GenServer.cast(topology_pid, {:server_description, server_description})
+        end
+        #require IEx; IEx.pry
+	#:debugger.start()
         case Error.should_retry_write(error, cmd, opts) do
           true  ->
           with :ok <- Session.select_server(session, opts) do
-           exec_command_session(session, cmd, Keyword.put(opts, :write_counter, 2))
+           exec_command_session(topology_pid, session, cmd, Keyword.put(opts, :write_counter, 2))
           end
           false -> {:error, error}
         end
@@ -748,13 +756,13 @@ defmodule Mongo do
   @spec exec_command(GenServer.server, BSON.document, Keyword.t) :: {:ok, BSON.document | nil} | {:error, Mongo.Error.t}
   def exec_command(conn, cmd, opts) do
     with {:ok, _cmd, {doc, event}} <- DBConnection.execute(conn, %Query{action: :command}, [cmd], defaults(opts)),
-         {:ok, doc} <- check_for_error(doc, event) do
+         {:ok, doc} <- check_for_error(doc, event, conn) do
       {:ok, doc}
     end
 
   end
 
-  defp check_for_error(%{"ok" => ok} = response, {event, duration}) when ok == 1 do
+  defp check_for_error(%{"ok" => ok} = response, {event, duration}, conn) when ok == 1 do
     Events.notify(%CommandSucceededEvent{
       reply: response,
       duration: duration,
@@ -765,9 +773,9 @@ defmodule Mongo do
     }, :commands)
     {:ok, response}
   end
-  defp check_for_error(doc, {event, duration}) do
+  defp check_for_error(doc, {event, duration}, address) do
 
-    error = Mongo.Error.exception(doc)
+    error = Mongo.Error.exception(doc, address)
 
     Events.notify(%CommandFailedEvent{
       failure: error,

lib/mongo/error.ex

@@ -2,7 +2,7 @@ defmodule Mongo.Error do
 
   alias Mongo.Events
 
-  defexception [:message, :code, :host, :error_labels, :resumable, :retryable_reads, :retryable_writes]
+  defexception [:message, :code, :host, :error_labels, :resumable, :retryable_reads, :retryable_writes, :not_master_or_recovering, :address]
 
   @host_unreachable                     6
   @host_not_found                       7
@@ -21,6 +21,18 @@ defmodule Mongo.Error do
   @stale_config                         13388
   @retry_change_stream                  234
   @failed_to_satisfy_read_preference    133
+  
+  # https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#not-master-and-node-is-recovering
+  @not_master_or_recovering [
+    @interrupted_at_shutdown,
+    @interrupted_due_to_repl_state_change,
+    @not_master_or_secondary,
+    @primary_stepped_down,
+    @shutdown_in_progress,
+    
+    @not_master,
+    @not_master_no_slaveok,
+  ]
 
   @retryable_writes [@interrupted_at_shutdown, @interrupted_due_to_repl_state_change, @not_master, @not_master_no_slaveok,
                      @not_master_or_secondary, @primary_stepped_down, @shutdown_in_progress, @host_not_found,
@@ -42,7 +54,8 @@ defmodule Mongo.Error do
     error_labels: [String.t] | nil,
     resumable: boolean,
     retryable_reads: boolean,
-    retryable_writes: boolean
+    retryable_writes: boolean,
+    not_master_or_recovering: boolean,
   }
 
   def message(e) do
@@ -60,12 +73,13 @@ defmodule Mongo.Error do
     %Mongo.Error{message: "#{host} ssl #{action}: #{formatted_reason} - #{inspect(reason)}", host: host, resumable: false}
   end
 
-  def exception(%{"code" => code, "errmsg" => msg} = doc) do
+  def exception(%{"code" => code, "errmsg" => msg} = doc, address) do
     errorLabels     = doc["errorLabels"] || []
     resumable       = Enum.any?(@resumable, &(&1 == code)) || Enum.any?(errorLabels, &(&1 == "ResumableChangeStreamError"))
     retryable_reads = Enum.any?(@retryable_reads, &(&1 == code)) || Enum.any?(errorLabels, &(&1 == "RetryableReadError"))
     retryable_writes = Enum.any?(@retryable_writes, &(&1 == code)) || Enum.any?(errorLabels, &(&1 == "RetryableWriteError"))
-    %Mongo.Error{message: msg, code: code, error_labels: errorLabels, resumable: resumable, retryable_reads: retryable_reads, retryable_writes: retryable_writes}
+    not_master_or_recovering = Enum.any?(@not_master_or_recovering, &(&1 == code))
+    %Mongo.Error{message: msg, code: code, error_labels: errorLabels, resumable: resumable, retryable_reads: retryable_reads, retryable_writes: retryable_writes, not_master_or_recovering: not_master_or_recovering, address: address}
   end
   def exception(message: message, code: code) do
     %Mongo.Error{message: message, code: code, resumable: Enum.any?(@resumable, &(&1 == code))}
@@ -76,7 +90,7 @@ defmodule Mongo.Error do
   end
 
   @doc """
-  Return true if the error is retryalble for read operations.
+  Return true if the error is retryable for read operations.
   """
   def should_retry_read(%Mongo.Error{retryable_reads: true}, cmd, opts) do
     [{command_name,_}|_] = cmd
@@ -93,7 +107,7 @@ defmodule Mongo.Error do
   end
 
   @doc """
-  Return true if the error is retryalble for writes operations.
+  Return true if the error is retryable for writes operations.
   """
   def should_retry_write(%Mongo.Error{retryable_writes: true}, cmd, opts) do
     [{command_name,_}|_] = cmd

lib/mongo/session.ex

@@ -473,7 +473,10 @@ defmodule Mongo.Session do
           |> ReadPreference.add_read_preference(opts)
           |> filter_nils()
 
-    {:keep_state_and_data, {:ok, conn, cmd}}
+    {host, port} = Mongo.MongoDBConnection.Utils.hostname_port(opts)
+    address = "#{host}:#{port}"
+
+    {:keep_state_and_data, {:ok, conn, cmd, address}}
   end
   def handle_call_event({:bind_session, cmd}, :starting_transaction,
         %Session{conn: conn,