defmodule HTTP2ConnectionFlowControlStallTest do
  # Reproduction of unbounded blocking on the *connection* send window.
  #
  # Bandit bounds a stream-window block with `stream.read_timeout` (15s), but a block on the
  # connection window is a `GenServer.call(..., :infinity)` into the connection process, whose
  # reply is queued in `Connection.pending_sends` until a connection-level WINDOW_UPDATE arrives.
  # Nothing bounds that wait, nothing cancels the queued send, and RST_STREAM cannot be observed
  # by a stream process that is blocked inside the call.

  use ExUnit.Case, async: true
  use ServerHelpers

  setup :https_server

  @cancel Bandit.HTTP2.Errors.cancel()

  describe "blocking on the connection send window" do
    test "is not cancelled by RST_STREAM, and stale data is sent after the stream is reset",
         context do
      context = blocking_server(context)
      socket = SimpleH2Client.setup_connection(context)

      plug_pid = start_blocked_stream(socket, context, grant: :stream_window)

      # The plug is blocked inside the connection process call, and the client cancels the stream
      SimpleH2Client.send_rst_stream(socket, 1, @cancel)
      Process.sleep(200)

      # The reset is queued in the stream process' mailbox but never acted on: the request keeps
      # its process, its Plug state and its queued body bytes alive
      assert Process.alive?(plug_pid)
      refute_received {:blocking_chunk_returned, ^plug_pid}

      # Growing the connection window drains pending_sends and puts the remainder of the body on
      # the wire for a stream the client already reset
      SimpleH2Client.send_window_update(socket, 0, 1_000_000)

      assert {:ok, 1, false, chunk} = SimpleH2Client.recv_body(socket)
      assert byte_size(chunk) == 4_465
      assert_receive {:blocking_chunk_returned, ^plug_pid}, 1_000
    end

    @tag :slow
    @tag timeout: 120_000
    test "is not bounded by any timeout while the connection is kept alive by PINGs", context do
      context = blocking_server(context)
      socket = SimpleH2Client.setup_connection(context)

      plug_pid = start_blocked_stream(socket, context, grant: :stream_window)

      # Well past the 15s that bounds a stream-window block
      ping_until(socket, 20_000)

      assert SimpleH2Client.connection_alive?(socket)
      assert Process.alive?(plug_pid)
      refute_received {:blocking_chunk_returned, ^plug_pid}
    end

    @tag :slow
    @tag timeout: 120_000
    test "contrast: blocking on the stream send window is bounded at 15s", context do
      context = blocking_server(context)
      socket = SimpleH2Client.setup_connection(context)

      plug_pid = start_blocked_stream(socket, context, grant: :connection_window)

      ping_until(socket, 16_000)

      assert_received {:blocking_chunk_error, ^plug_pid,
                       "Timeout waiting for space in the send_window"}

      refute Process.alive?(plug_pid)
    end
  end

  defp blocking_server(context) do
    context
    |> https_server(
      plug: blocking_plug(self()),
      thousand_island_options: [read_timeout: 2_000]
    )
    |> Enum.into(context)
  end

  # Writes 7 x 10k chunks against a 65_535 byte window, so the 7th chunk cannot be fully sent
  defp blocking_plug(test_pid) do
    fn conn, _opts ->
      data = String.duplicate("a", 10_000)

      send(test_pid, {:plug_ready, self()})
      receive do: (:go -> :ok)

      conn = send_chunked(conn, 200)

      conn =
        Enum.reduce(1..6, conn, fn _i, conn ->
          {:ok, conn} = chunk(conn, data)
          conn
        end)

      send(test_pid, {:blocking_chunk_started, self()})

      case chunk(conn, data) do
        {:ok, conn} ->
          send(test_pid, {:blocking_chunk_returned, self()})
          conn

        {:error, reason} ->
          send(test_pid, {:blocking_chunk_error, self(), reason})
          conn
      end
    end
  end

  # Drives the plug until it is blocked with 4_465 bytes of body still unsent, leaving whichever
  # window was not granted as the one that blocks
  defp start_blocked_stream(socket, context, grant: grant) do
    SimpleH2Client.send_simple_headers(socket, 1, :get, "/blocking_test", context.port)

    assert_receive {:plug_ready, plug_pid}, 1_000

    case grant do
      :stream_window -> SimpleH2Client.send_window_update(socket, 1, 1_000_000)
      :connection_window -> SimpleH2Client.send_window_update(socket, 0, 1_000_000)
    end

    # The window update is only observable once the plug produces frames, so let the connection
    # process apply it before any body is written
    Process.sleep(100)
    send(plug_pid, :go)

    SimpleH2Client.successful_response?(socket, 1, false)
    Enum.each(1..6, fn _i -> SimpleH2Client.recv_body(socket) end)
    assert {:ok, 1, false, chunk} = SimpleH2Client.recv_body(socket)
    assert byte_size(chunk) == 5_535

    assert_receive {:blocking_chunk_started, ^plug_pid}, 1_000
    refute_receive {:blocking_chunk_returned, ^plug_pid}, 500

    plug_pid
  end

  # Send-only keepalive: PING ACKs are left in the socket buffer so this never consumes the
  # frames a test is asserting on
  defp ping_until(socket, duration) do
    deadline = System.monotonic_time(:millisecond) + duration

    Stream.repeatedly(fn ->
      SimpleH2Client.send_frame(socket, 6, 0, 0, <<1, 2, 3, 4, 5, 6, 7, 8>>)
      Process.sleep(250)
      System.monotonic_time(:millisecond)
    end)
    |> Enum.find(&(&1 >= deadline))
  end
end