Sending a zip created on-the-fly via Plug.Conn.chunk fails after around 200Mb

[sezaru]

I'm having problem when using Plug.Conn.chunk/2 to send a stream of a zip file generated on-the-fly in Chrome. I'm not sure if I'm doing something wrong, but all the examples I see seem pretty similar to mine.

Here is my controller module:

defmodule CoreWeb.Export.AppointmentImagesController do
  @moduledoc false

  alias Core.Marketplace.Ap.AppointmentEvent

  use CoreWeb, :controller

  def index(conn, %{"appointment_id" => appointment_id} = _params) do
    %{current_user: actor} = conn.assigns

    case Ash.get(AppointmentEvent, %{id: appointment_id}, actor: actor) do
      {:ok, appointment} ->
        IO.puts("generating stream")

        stream = image_as_zip_stream!(appointment)

        IO.puts("start download")

        conn =
          conn
          |> put_resp_content_type("application/zip")
          |> put_resp_header("content-disposition", ~s[attachment; filename="#{appointment_id}.zip"])
          |> send_chunked(:ok)

        IO.puts("start sending chunks")

        stream
        |> Stream.each(fn results ->
            case Plug.Conn.chunk(conn, results) do
              {:ok, _} -> IO.puts("got here!!!")
              {:error, error} -> dbg(error)
            end
          end)
        |> Stream.run()

        IO.puts("DONE!!!")

        conn

      {:error, _} ->
        conn
        |> put_status(500)
        |> json(%{error: "Failed to generete the zip file, try again later."})
    end
  end

  defp image_as_zip_stream!(appointment) do
    appointment
    |> Map.fetch!(:event_json)
    |> Map.fetch!("synced")
    |> Enum.with_index()
    |> Enum.map(fn {%{"s3Url" => url}, index} ->
      [bucket, path] = url |> URI.parse() |> Map.fetch!(:path) |> String.split("/", parts: 2, trim: true)

      bucket
      |> ExAws.S3.download_file(path, "#{index}.jpg")
      |> ExAws.stream!()
      |> then(& Zstream.entry("#{index}.jpg", &1))
    end)
    |> Zstream.zip()
  end
end

As you can see, I get some images, create a stream that generates a zip from it and send via Plug.Conn.chunk/2.

When I request this from chrome, the download will start and it will show the download as resuming:

When the download is around 200mb, the download will fail and I will get a Network Error in chrome:

I also do not get any error message in the backend, it will just stop sending chunks:

[josevalim]

It is hard to say what is the root cause. At first, can you include versions for Erlang, Elixir, Plug and the Web Server you are using (Cowboy, Bandit)?

I also believe you are supposed to retain the connection returned from chunk/2 but you are discarding it. Check out the example in the docs: https://hexdocs.pm/plug/Plug.Conn.html#chunk/2

[josevalim]

Also try downloading it on other browsers and on curl, to rule out it being a Chrome issue.

[sezaru]

Sorry, I forgot to add the libraries versions:

erlang: 26.2.1
elixir: 1.16.2
plug_cowboy: 2.7.0

I was not using the Enum.reduce_while because I thought it would eagerly consume the stream before sending the data, but it doesn't seem to do this, so I changed to it and started using the same conn:

        conn =
          stream
          |> Enum.reduce_while(conn, fn chunk, conn ->
            case Plug.Conn.chunk(conn, chunk) do
              {:ok, conn} ->
                IO.puts("got here!!!")

                {:cont, conn}

              _ ->
                {:halt, conn}
            end
          end)

I then tried again in Chrome but got the same error. Firefox also fails, and curl gives me the following error:

~ » curl "http://localhost:4000/export/appointment_images?appointment_id=aea60d76-0172-4c5a-bba0-2afb5aa46ad0" --output my.zip
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  118M    0  118M    0     0  2024k      0 --:--:--  0:01:00 --:--:-- 6709k
curl: (18) transfer closed with outstanding read data remaining

[josevalim]

I cannot reproduce it:

Mix.install([:plug, :plug_cowboy])

defmodule MyPlug do
  import Plug.Conn

  def init(options) do
    # initialize options
    options
  end

  def call(conn, _opts) do
    conn =
      conn
      |> put_resp_content_type("text/plain")
      |> send_chunked(200)

    ["aaaaaaaaa", "bbbbbbbbbb", "ccccccccccc"]
    |> Enum.map(&String.duplicate(&1, 100))
    |> Stream.cycle()
    |> Stream.take(1_000_000)
    |> Enum.reduce_while(conn, fn chunk, conn ->
      case Plug.Conn.chunk(conn, chunk) do
        {:ok, conn} ->
          {:cont, conn}

        {:error, :closed} ->
          {:halt, conn}
      end
    end)
  end
end

require Logger
webserver = {Plug.Cowboy, plug: MyPlug, scheme: :http, options: [port: 4000]}
{:ok, _} = Supervisor.start_link([webserver], strategy: :one_for_one)
Logger.info("Plug now running on localhost:4000")
Process.sleep(:infinity)

followed by curl localhost:4000 --output foo streams 1GB almost immediately. I also tried smaller chunks by removing String.duplicate and it worked just as fine.

[sezaru]

Thanks for the help so far.

Tomorrow I will try to create a small project that reproduces the issue and post here

[sezaru]

So, I changed your script to something that I think is more similar to what my code is doing to trigger the issue:

Mix.install([:plug, :plug_cowboy, :zstream])

defmodule MyPlug do
  import Plug.Conn

  def init(options) do
    # initialize options
    options
  end

  def call(conn, _opts) do
    conn =
      conn
      |> put_resp_content_type("text/plain")
      |> send_chunked(200)

    # Fake 5MB binary representing some image
    fake_binary = :crypto.strong_rand_bytes(5242880)

    1..150
    |> Enum.map(fn index ->
      {:ok, pid} = StringIO.open(fake_binary)

      stream = IO.binstream(pid, :line) |> Stream.map(fn chunk ->
        Process.sleep(1)

        chunk
      end)

      Zstream.entry("#{index}.jpg", stream)
    end)
    |> Zstream.zip()
    |> Enum.reduce_while(conn, fn chunk, conn ->
      case Plug.Conn.chunk(conn, chunk) do
        {:ok, conn} ->
          {:cont, conn}

        {:error, :closed} ->
          {:halt, conn}
      end
    end)
  end
end

require Logger
webserver = {Plug.Cowboy, plug: MyPlug, scheme: :http, options: [port: 4000]}
{:ok, _} = Supervisor.start_link([webserver], strategy: :one_for_one)
Logger.info("Plug now running on localhost:4000")
Process.sleep(:infinity)

As you can see, I created a random binary with around 5Mb in size since that is the mean size of the images I'm putting into the zip file.

Then, for each entry in the zip, I added a sleep, the reason for that is that I download each image from S3 in a stream in my code, and that, of course, creates some latency, so I'm using Process.sleep(1) to simulate that, with these changes, I was able to trigger the same issue.

[josevalim]

Thank you. I was able to reproduce the issue with Cowboy but I could not reproduce it with Bandit. This makes me believe this is not a Plug issue, but rather a web server issue. You can try Bandit in your application OR isolate this report directly on top of Erlang+Cowboy and file it upstream (if doing so, I'd start by first reproducing the issue without zstream).

[sezaru]

using Bandit worked great for me, thanks! I will try my hands in Erlang+Cowboy to create a small example that triggers the issue and create a new issue in their repo.