How do you crash bad connections?

[16ajbm]

When starting a connection under the application supervision tree as {Mongo, opts}, bad connections aren't crashing the process. Even when manipulating any of the timeouts (:checkout_timeout, :connect_timeout, etc.) it seems to sit forever logging [error] Mongo.MongoDBConnection ... :econnrefused. When manually attempting to create a cursor via iex on a bad connection, eventually there is a message timeout at 60 seconds.

Is there a way to crash this harder and faster? I would like to bring down the whole application when there is connection issues, that way I can rely on K8s to restart things.

[zookzook]

Is there a way to crash this harder and faster? I would like to bring down the whole application when there are connection issues, that way I can rely on K8s to restart things.

Keep in mind, that erlang/elixir is a cluster system. If you using Kubernetes, there is no reason to restart the whole pod/container application, because the erlang pod will try to restart the process that handles the connection. Because of connection issues, there is no need to restart the erlang/elixir pod. The process itself is restarted and the new process will try to connect. And what would a restart of a pod change, when the connection can not be established? You will see a lot of restarts of pods.

[zookzook]

Can you provide an example, of when the cursor is stalling? If you use a cursor then a request is sent to the server and if there is no connection then you will get a timeout at some point.

[16ajbm]

Sure. I would like to consume the via the Flow library. I saw in the examples (Changestream specifically) the method of using spawn() and Enum.each(). Is the timeout from the cursor configurable?

defmodule MongoConnectionTest.Flow do
  use Flow

  @collection "collection"

  require Logger

  def start_link(_) do
    opts = []
    
    # timeout occurs here
    cursor = get_cursor(%{last_resume_token: nil})

    cursor
    |> Flow.from_enumerable(max_demand: 1)
    |> Flow.map_batch(fn doc -> preprocess(doc) end)
    |> Flow.map(fn doc -> process(doc) end)
    |> Flow.start_link(opts)
  end


 # taken from changestream example
  defp get_cursor(%{last_resume_token: nil}) do
    Mongo.watch_collection(:mongo_test, @collection, [], fn token -> token end,
      max_time: 2_000
    )
  end

  defp get_cursor(%{last_resume_token: token}) do
    Mongo.watch_collection(:mongo_test, @collection, [], fn token -> token end,
      max_time: 2_000,
      resume_after: token
    )
  end
end

[zookzook]

What is the task of using Flow here?

[16ajbm]

For batch processing of related documents. Sorry, the example was not exactly reflective of that originally. In the actual implementation there are more stages, just omitted for simplicity.

[zookzook]

The watch_collection uses a pulling approach. After a timeout, it sends a request to MongoDB again. If the connection could not be established an error is returned. So, try your application without Flow to see if something is stalling.