fix a crash in the streaming hello monitor, if the server sends more than one response at once

Author: zookzook

CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## 0.9.2 (2022-09-07)
+* Bugfix
+  * fix a crash in the streaming hello monitor, if the server sends more than one response at once 
+
 ## 0.9.1 (2022-05-27)
 * Bugfix
   * add backward compatible for Elixir < 1.13 (thanks to ja-jimenez)

lib/mongo/messages.ex

@@ -39,7 +39,7 @@ defmodule Mongo.Messages do
 
   @header_size 4 * 4
 
-  defrecordp :msg_header, [:length, :request_id, :response_to, :op_code]
+  defrecord :msg_header, [:length, :request_id, :response_to, :op_code]
   defrecord :op_query, [:flags, :coll, :num_skip, :num_return, :query, :select]
   defrecord :op_reply, [:flags, :cursor_id, :from, :num, :docs]
   defrecord :sequence, [:size, :identifier, :docs]
@@ -51,16 +51,20 @@ defmodule Mongo.Messages do
     Decodes the header from response of a request sent by the mongodb server
   """
   def decode_header(iolist) when is_list(iolist) do
-    case IO.iodata_length(iolist) >= @header_size do
-      true -> iolist |> IO.iodata_to_binary() |> decode_header()
-      false -> :error
+    case IO.iodata_length(iolist) == @header_size do
+      true ->
+        iolist
+        |> IO.iodata_to_binary()
+        |> decode_header()
+
+      false ->
+        :error
     end
   end
 
-  def decode_header(<<length::int32, request_id::int32, response_to::int32, op_code::int32, rest::binary>>) do
-    ## todo don't subtract header-size here
+  def decode_header(<<length::int32, request_id::int32, response_to::int32, op_code::int32>>) do
     header = msg_header(length: length - @header_size, request_id: request_id, response_to: response_to, op_code: op_code)
-    {:ok, header, rest}
+    {:ok, header}
   end
 
   def decode_header(_binary), do: :error

lib/mongo/version.ex

@@ -35,7 +35,15 @@ defmodule Mongo.Version do
     # Supports features available from 5.0 and onwards.
     wire_version_50: 13,
     # Supports features available from 5.1 and onwards.
-    wire_version_51: 14
+    wire_version_51: 14,
+    # Supports features available from 5.2 and onwards.
+    wire_version_52: 15,
+    # Supports features available from 5.3 and onwards.
+    wire_version_53: 16,
+    # Supports features available from 6.0 and onwards.
+    wire_version_60: 17,
+    # Supports features available from 6.1 and onwards.
+    wire_version_61: 18
   ]
 
   for {key, value} <- values do

lib/mongo_db_connection/utils.ex

@@ -12,6 +12,8 @@ defmodule Mongo.MongoDBConnection.Utils do
   # currently not used @reply_shard_config_stale 0x4
   # currently not used @reply_await_capable      0x8
 
+  @header_size 4 * 4
+
   @doc """
     Sends a request id and waits for the response with the same id
   """
@@ -97,22 +99,24 @@ defmodule Mongo.MongoDBConnection.Utils do
   end
 
   defp recv_data(nil, "", %{connection: {mod, socket}} = state) do
-    case mod.recv(socket, 0, state.timeout) do
+    case mod.recv(socket, @header_size, state.timeout) do
       {:ok, tail} -> recv_data(nil, tail, state)
       {:error, reason} -> recv_error(reason, state)
     end
   end
 
   defp recv_data(nil, data, %{connection: {mod, socket}} = state) do
     case decode_header(data) do
-      {:ok, header, rest} ->
-        recv_data(header, rest, state)
+      {:ok, header} ->
+        msg_header(length: length) = header
 
-      :error ->
-        case mod.recv(socket, 0, state.timeout) do
-          {:ok, tail} -> recv_data(nil, [data | tail], state)
+        case mod.recv(socket, length, state.timeout) do
+          {:ok, data} -> recv_data(header, data, state)
           {:error, reason} -> recv_error(reason, state)
         end
+
+      :error ->
+        recv_error(:header_not_found, state)
     end
   end
 
@@ -122,6 +126,7 @@ 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)

test/mongo/transaction_retries_test.exs

@@ -165,6 +165,6 @@ defmodule Mongo.TransactionRetriesTest do
     catcher
     |> EventCatcher.succeeded_events()
     |> Enum.map(fn event -> event.command_name end)
-    |> Enum.reject(fn event -> event == :isMaster end)
+    |> Enum.reject(fn event -> event == :isMaster || event == :more_to_come end)
   end
 end