I’ve spent some more time on this and dropped down to Mint to have a better understanding of what’s happening. The TransportError seems to be related to the specific server that I was testing with which I now replaced with a locally running Phoenix app.

To recap: I’m sending out emails via REST, so I need to send quite a lot of JSON requests that share a big binary blob (bese-64 encoded attachment). I’m caching the blob, but I’m still facing memory usage spikes which in the end causes out of memory issues. Here’s the test script that I’m using to reproduce this:

Mix.install([:jason, :mint])
:observer.start()

defmodule TestMod do
  @scheme :https
  @host "localhost"
  @port 4040
  @method "POST"
  @path "/test"
  @headers [{"content-type", "application/json"}]
  @sleep 3000
  # Don't complain on a self-signed certificat
  @connect_opts [transport_opts: [verify: :verify_none]]

  def request_regular() do
    IO.puts("Loading data...")
    body = big_json("regular")
    Process.sleep(@sleep)
    IO.puts("Regular request...")
    {:ok, conn} = Mint.HTTP1.connect(@scheme, @host, @port, @connect_opts)
    {:ok, conn, request_ref} = Mint.HTTP1.request(conn, @method, @path, @headers, body)
    receive_response(conn, request_ref)
  end

  def request_stream() do
    IO.puts("Loading data...")
    body = big_json("stream")
    Process.sleep(@sleep)
    IO.puts("Streaming request body...")
    {:ok, conn} = Mint.HTTP1.connect(@scheme, @host, @port, @connect_opts)
    {:ok, conn, request_ref} = Mint.HTTP1.request(conn, @method, @path, @headers, :stream)
    {:ok, conn} = Mint.HTTP1.stream_request_body(conn, request_ref, body)
    {:ok, conn} = Mint.HTTP1.stream_request_body(conn, request_ref, :eof)
    receive_response(conn, request_ref)
  end

  def request_stream_chunked_iodata() do
    IO.puts("Loading data...")
    body = big_json("chunked iodata")
    Process.sleep(@sleep)
    IO.puts("Streaming request body as chunked iodata...")
    {:ok, conn} = Mint.HTTP1.connect(@scheme, @host, @port, @connect_opts)
    {:ok, conn, request_ref} = Mint.HTTP1.request(conn, @method, @path, @headers, :stream)
    chunked = chunk_iodata(body)
    {:ok, conn} = Mint.HTTP1.stream_request_body(conn, request_ref, chunked)
    {:ok, conn} = Mint.HTTP1.stream_request_body(conn, request_ref, :eof)
    receive_response(conn, request_ref)
  end

  def request_stream_chunked_binary() do
    IO.puts("Loading data...")
    body = big_json("chunked binaries")
    Process.sleep(@sleep)
    IO.puts("Streaming request body in binary chunks...")
    {:ok, conn} = Mint.HTTP1.connect(@scheme, @host, @port, @connect_opts)
    {:ok, conn, request_ref} = Mint.HTTP1.request(conn, @method, @path, @headers, :stream)
    chunks = iodata_to_binary_chunks(body, [])

    {:ok, conn} =
      Enum.reduce(chunks, {:ok, conn}, fn chunk, {:ok, conn} ->
        Mint.HTTP1.stream_request_body(conn, request_ref, chunk)
      end)

    {:ok, conn} = Mint.HTTP1.stream_request_body(conn, request_ref, :eof)
    receive_response(conn, request_ref)
  end

  def receive_response(conn, request_ref, result \\ %{}) do
    receive do
      message ->
        case Mint.HTTP1.stream(conn, message) do
          :unknown ->
            IO.inspect(message, label: ">>> unknown message")

          {:ok, conn, responses} ->
            reduced =
              Enum.reduce(responses, result, fn response, result ->
                case response do
                  {:status, _request_ref, status} ->
                    Map.put(result, :status, status)

                  {:headers, _request_ref, headers} ->
                    Map.put(result, :headers, headers)

                  {:data, _request_ref, body_chunk} ->
                    Map.update(result, :body, body_chunk, fn body ->
                      body <> body_chunk
                    end)

                  {:done, _request_ref} ->
                    {:ok, result}

                  {:error, _request_ref, reason} ->
                    {:error, reason}
                end
              end)

            case reduced do
              {:ok, result} -> {:ok, result}
              {:error, reason} -> {:error, reason}
              result -> receive_response(conn, request_ref, result)
            end
        end
    end
  end

  def big_binary(megabytes \\ 10) do
    String.duplicate("-", 1024 * 1024 * megabytes)
  end

  def big_json(foo, megabytes \\ 10) do
    Jason.encode_to_iodata!(%{content: big_binary(megabytes), foo: foo})
  end

  def iodata_to_binary_chunks([], acc) do
    Enum.reverse(acc)
  end

  def iodata_to_binary_chunks(list, acc) do
    case hd(list) do
      number when is_integer(number) ->
        iodata_to_binary_chunks(tl(list), [<<number>> | acc])

      binary when is_binary(binary) ->
        iodata_to_binary_chunks(tl(list), chunk_binary(binary, acc))

      [] ->
        iodata_to_binary_chunks(tl(list), acc)

      nested when is_list(nested) ->
        iodata_to_binary_chunks([hd(nested), tl(nested) | tl(list)], acc)
    end
  end

  @binary_chunk_size 102_400

  def chunk_binary(str, acc) do
    case String.split_at(str, @binary_chunk_size) do
      {slice, ""} -> [slice | acc]
      {slice, rest} -> chunk_binary(rest, [slice | acc])
    end
  end

  def chunk_iodata([]) do
    []
  end

  def chunk_iodata(list) when is_list(list) do
    [chunk_iodata(hd(list)) | chunk_iodata(tl(list))]
  end

  def chunk_iodata(binary) when is_binary(binary) do
    chunk_binary(binary, []) |> Enum.reverse()
  end

  def chunk_iodata(other) do
    other
  end
end

I’m testing the following approaches:

1. Sending iodata,
2. Streaming all iodata at once,
3. Streaming all iodata but chunking big binaries into smaller ones,
4. Streaming in chunks of binaries: converting iodata to a list of binary chunks where each chunk has a max size limit.

Screenshot 2023-03-05 at 21.50.581924×1424 135 KB

It looks like only the last approach avoids memory spikes. Since it seems the problem only shows up with HTTPS and not HTTP, I’m guessing this has something to do with how the data is encrypted by the :ssl module. My working theory is that the last approach fits nicely with the encryption algorithm. But I’d prefer for the chunking to happen somewhere in the library code, not mine. Otherwise it feels like the solution is a bit magical and quite fragile.

@ericmj @whatyouhide @voltone I’d appreciate if you could chime in on this.

Suggestion: Add a link from the Erlang Forum thread to this thread. This allows people to read both threads and avoid duplicate work.