Batch HTTP/2 receive-window refills

Previously `refill_client_windows/3` sent a WINDOW_UPDATE on both the
connection and the stream after every DATA frame, with the increment
set to the frame's byte size. That kept the advertised window pinned
at its peak but tied outbound WINDOW_UPDATE traffic one-to-one with
inbound DATA frames.

An adversarial server can exploit that ratio. By sending many small
DATA frames — in the limit, one byte of body per frame — it can force
the client to emit one 13-byte WINDOW_UPDATE per frame. At high frame
rates that's a small but real client-side amplification: a flood of
outbound control frames driven entirely by the peer.

This change gates refills on a threshold. The client tracks the
current remaining window for the connection and each stream and only
sends a WINDOW_UPDATE once that remaining drops to
`:receive_window_update_threshold` bytes. The update then tops the
window straight back up to its configured peak. One frame per
`receive_window_size - receive_window_update_threshold` bytes
consumed, not per DATA frame. The default threshold is 160_000 bytes
— roughly 10× the default 16 KB max frame size, leaving the server a
safety margin before the window would starve it.

Behaviour-wise:

  * With the 4 MB / 16 MB default windows, the client sends roughly
    one stream-level WINDOW_UPDATE per ~3.84 MB consumed (previously
    ~250 per 4 MB), and one connection-level update per ~15.84 MB
    (previously ~1000 per 16 MB).
  * Callers that explicitly set the stream or connection window down
    to the 65_535 spec minimum get the old behaviour — one refill per
    frame — because remaining is always below the default 160_000
    threshold.

The threshold is tunable via the new `:receive_window_update_threshold`
option to `Mint.HTTP.connect/4`.

Author: ericmj

lib/mint/http.ex

@@ -243,6 +243,12 @@ defmodule Mint.HTTP do
       on stream 0 as part of the connection preface. Defaults to 16 MB. Can be
       raised later with `Mint.HTTP2.set_window_size/3`.
 
+    * `:receive_window_update_threshold` - (integer) the minimum number of bytes of receive
+      window that must remain on a connection or stream before a `WINDOW_UPDATE`
+      frame is sent to refill it. Lower values send more frequent, smaller updates;
+      higher values batch updates into fewer, larger ones. Defaults to 160_000
+      (approximately 10× the default max frame size).
+
   There may be further protocol specific options that only take effect when the corresponding
   connection is established. Check `Mint.HTTP1.connect/4` and `Mint.HTTP2.connect/4` for
   details.

lib/mint/http2.ex

@@ -147,6 +147,12 @@ defmodule Mint.HTTP2 do
   @default_stream_window_size 4 * 1024 * 1024
   @max_window_size 2_147_483_647
 
+  # Defer refilling the receive window until it has dropped to this many
+  # bytes — roughly 10× the default 16 KB max frame size, so the server
+  # has a safety margin before the window would starve it. See
+  # `refill_client_windows/3`.
+  @default_receive_window_update_threshold 160_000
+
   @default_max_frame_size 16_384
   @valid_max_frame_size_range @default_max_frame_size..16_777_215
 
@@ -189,6 +195,15 @@ defmodule Mint.HTTP2 do
     # matching WINDOW_UPDATE to bring the server's view from the spec
     # default of 65_535 up to the advertised peak.
     receive_window_size: @default_window_size,
+    # `receive_window` is the server's current view of our receive
+    # window — decremented by DATA frame sizes as they arrive, bumped
+    # back up to `receive_window_size` whenever we send a
+    # WINDOW_UPDATE. When it drops to `receive_window_update_threshold`, we
+    # refill it back to the peak in one frame.
+    receive_window: @default_window_size,
+    # Minimum remaining receive window before we send a WINDOW_UPDATE.
+    # Configurable via the `:receive_window_update_threshold` connect option.
+    receive_window_update_threshold: @default_receive_window_update_threshold,
     encode_table: HPAX.new(4096),
     decode_table: HPAX.new(4096),
 
@@ -896,7 +911,8 @@ defmodule Mint.HTTP2 do
 
   def set_window_size(%__MODULE__{} = conn, :connection, new_size) do
     do_set_window_size(conn, 0, conn.receive_window_size, new_size, fn conn, size ->
-      put_in(conn.receive_window_size, size)
+      conn = put_in(conn.receive_window_size, size)
+      put_in(conn.receive_window, size)
     end)
   catch
     :throw, {:mint, conn, error} -> {:error, conn, error}
@@ -908,7 +924,8 @@ defmodule Mint.HTTP2 do
         current = conn.streams[stream_id].receive_window_size
 
         do_set_window_size(conn, stream_id, current, new_size, fn conn, size ->
-          put_in(conn.streams[stream_id].receive_window_size, size)
+          conn = put_in(conn.streams[stream_id].receive_window_size, size)
+          put_in(conn.streams[stream_id].receive_window, size)
         end)
 
       :error ->
@@ -1104,12 +1121,10 @@ defmodule Mint.HTTP2 do
     scheme_string = Atom.to_string(scheme)
     mode = Keyword.get(opts, :mode, :active)
     log? = Keyword.get(opts, :log, false)
-
-    connection_window_size =
-      Keyword.get(opts, :connection_window_size, @default_connection_window_size)
-
+    connection_window_size = Keyword.get(opts, :connection_window_size, @default_connection_window_size)
     validate_window_size!(:connection_window_size, connection_window_size)
-
+    receive_window_update_threshold = Keyword.get(opts, :receive_window_update_threshold, @default_receive_window_update_threshold)
+    validate_receive_window_update_threshold!(receive_window_update_threshold)
     client_settings_params = Keyword.get(opts, :client_settings, [])
 
     client_settings_params =
@@ -1144,7 +1159,9 @@ defmodule Mint.HTTP2 do
       scheme: scheme_string,
       state: :handshaking,
       log: log?,
-      receive_window_size: connection_window_size
+      receive_window_size: connection_window_size,
+      receive_window: connection_window_size,
+      receive_window_update_threshold: receive_window_update_threshold
     }
 
     preface = build_preface(client_settings_params, connection_window_size)
@@ -1182,6 +1199,14 @@ defmodule Mint.HTTP2 do
     end
   end
 
+  defp validate_receive_window_update_threshold!(value) do
+    unless is_integer(value) and value >= 1 and value <= @max_window_size do
+      raise ArgumentError,
+            "the :receive_window_update_threshold option must be a positive integer no larger than " <>
+              "#{@max_window_size}, got: #{inspect(value)}"
+    end
+  end
+
   @doc """
   See `Mint.HTTP.get_socket/1`.
   """
@@ -1259,6 +1284,9 @@ defmodule Mint.HTTP2 do
       # SETTINGS_INITIAL_WINDOW_SIZE; can be bumped per-stream with
       # `set_window_size/3`.
       receive_window_size: conn.client_settings.initial_window_size,
+      # Current remaining receive window for this stream, tracked
+      # independently from the peak so that refills can be batched.
+      receive_window: conn.client_settings.initial_window_size,
       received_first_headers?: false
     }
 
@@ -1753,17 +1781,79 @@ defmodule Mint.HTTP2 do
     end
   end
 
+  # Accounts for `data_size` bytes arriving on the connection and on
+  # `stream_id`. Sends a WINDOW_UPDATE for either window only once its
+  # remaining receive credit drops to `conn.receive_window_update_threshold`;
+  # previously we sent one per DATA frame, so an adversarial server
+  # emitting many small frames could amplify its inbound bytes into a
+  # WINDOW_UPDATE flood of outbound frames. Batching caps that ratio at
+  # roughly one update per `(receive_window_size - threshold)` bytes
+  # consumed.
   defp refill_client_windows(conn, stream_id, data_size) do
-    connection_frame = window_update(stream_id: 0, window_size_increment: data_size)
-    stream_frame = window_update(stream_id: stream_id, window_size_increment: data_size)
+    conn = update_in(conn.receive_window, &(&1 - data_size))
 
-    if open?(conn) do
-      send!(conn, [Frame.encode(connection_frame), Frame.encode(stream_frame)])
+    conn =
+      case Map.fetch(conn.streams, stream_id) do
+        {:ok, _stream} ->
+          update_in(conn.streams[stream_id].receive_window, &(&1 - data_size))
+
+        :error ->
+          conn
+      end
+
+    frames =
+      []
+      |> maybe_refill_stream(conn, stream_id)
+      |> maybe_refill_conn(conn)
+
+    if frames != [] and open?(conn) do
+      conn = send!(conn, Enum.map(frames, &Frame.encode/1))
+      apply_refills(conn, frames)
     else
       conn
     end
   end
 
+  defp maybe_refill_conn(frames, conn) do
+    if conn.receive_window <= conn.receive_window_update_threshold do
+      increment = conn.receive_window_size - conn.receive_window
+      [window_update(stream_id: 0, window_size_increment: increment) | frames]
+    else
+      frames
+    end
+  end
+
+  defp maybe_refill_stream(frames, conn, stream_id) do
+    case Map.fetch(conn.streams, stream_id) do
+      {:ok, stream} ->
+        if stream.receive_window <= conn.receive_window_update_threshold do
+          increment = stream.receive_window_size - stream.receive_window
+
+          [
+            window_update(stream_id: stream_id, window_size_increment: increment) | frames
+          ]
+        else
+          frames
+        end
+
+      :error ->
+        frames
+    end
+  end
+
+  defp apply_refills(conn, frames) do
+    Enum.reduce(frames, conn, fn
+      window_update(stream_id: 0), conn ->
+        put_in(conn.receive_window, conn.receive_window_size)
+
+      window_update(stream_id: stream_id), conn ->
+        put_in(
+          conn.streams[stream_id].receive_window,
+          conn.streams[stream_id].receive_window_size
+        )
+    end)
+  end
+
   # HEADERS
 
   defp handle_headers(conn, frame, responses) do
@@ -2108,6 +2198,8 @@ defmodule Mint.HTTP2 do
       ref: make_ref(),
       state: :reserved_remote,
       send_window_size: conn.server_settings.initial_window_size,
+      receive_window_size: conn.client_settings.initial_window_size,
+      receive_window: conn.client_settings.initial_window_size,
       received_first_headers?: false
     }
 

test/mint/http2/conn_test.exs

@@ -141,10 +141,12 @@ defmodule Mint.HTTP2Test do
          %{conn: conn} do
       assert HTTP2.get_window_size(conn, :connection) == 65_535
       assert conn.receive_window_size == 65_535
+      assert conn.receive_window == 65_535
 
       assert {:ok, conn} = HTTP2.set_window_size(conn, :connection, 1_000_000)
 
       assert conn.receive_window_size == 1_000_000
+      assert conn.receive_window == 1_000_000
 
       assert_recv_frames [
         window_update(stream_id: 0, window_size_increment: 934_465)
@@ -157,10 +159,12 @@ defmodule Mint.HTTP2Test do
       assert_recv_frames [headers(stream_id: stream_id)]
 
       current = conn.streams[stream_id].receive_window_size
+      assert conn.streams[stream_id].receive_window == current
 
       assert {:ok, conn} = HTTP2.set_window_size(conn, {:request, ref}, current + 10_000)
 
       assert conn.streams[stream_id].receive_window_size == current + 10_000
+      assert conn.streams[stream_id].receive_window == current + 10_000
 
       assert_recv_frames [
         window_update(stream_id: ^stream_id, window_size_increment: 10_000)
@@ -1874,6 +1878,7 @@ defmodule Mint.HTTP2Test do
       # a WINDOW_UPDATE for `16 MB - 65_535` so the server sees the peak
       # from the start.
       assert conn.receive_window_size == 16 * 1024 * 1024
+      assert conn.receive_window == 16 * 1024 * 1024
     end
 
     test "advertises the configured stream window via SETTINGS", %{conn: conn} do
@@ -1884,11 +1889,13 @@ defmodule Mint.HTTP2Test do
       assert_recv_frames [headers(stream_id: stream_id)]
 
       assert conn.streams[stream_id].receive_window_size == 4 * 1024 * 1024
+      assert conn.streams[stream_id].receive_window == 4 * 1024 * 1024
     end
 
     @tag connect_options: [connection_window_size: 1_000_000]
     test "supports a custom :connection_window_size", %{conn: conn} do
       assert conn.receive_window_size == 1_000_000
+      assert conn.receive_window == 1_000_000
     end
 
     @tag connect_options: [connection_window_size: 65_535]
@@ -1897,6 +1904,7 @@ defmodule Mint.HTTP2Test do
       # At 65_535 there's nothing to advertise beyond SETTINGS — the
       # preface should not carry an extra WINDOW_UPDATE.
       assert conn.receive_window_size == 65_535
+      assert conn.receive_window == 65_535
     end
 
     test "rejects a :connection_window_size below the spec minimum" do
@@ -1910,6 +1918,85 @@ defmodule Mint.HTTP2Test do
         HTTP2.initiate(:https, self(), "localhost", 443, connection_window_size: 2_147_483_648)
       end
     end
+
+    test "rejects a non-positive :receive_window_update_threshold" do
+      assert_raise ArgumentError, ~r/:receive_window_update_threshold/, fn ->
+        HTTP2.initiate(:https, self(), "localhost", 443, receive_window_update_threshold: 0)
+      end
+    end
+  end
+
+  describe "receive window batching" do
+    @describetag connect_options: [
+                   connection_window_size: 100_000,
+                   receive_window_update_threshold: 40_000,
+                   client_settings: [initial_window_size: 100_000]
+                 ]
+
+    test "does not send WINDOW_UPDATE until remaining window drops below threshold",
+         %{conn: conn} do
+      {conn, _ref} = open_request(conn)
+      assert_recv_frames [headers(stream_id: stream_id)]
+
+      # 50_000 bytes consumed leaves 50_000 remaining on both windows,
+      # above the 40_000 threshold, so no WINDOW_UPDATE should go out.
+      chunk = String.duplicate("a", 10_000)
+
+      frames = for _ <- 1..5, do: data(stream_id: stream_id, data: chunk)
+
+      assert {:ok, %HTTP2{} = _conn, _responses} = stream_frames(conn, frames)
+
+      assert_recv_frames []
+    end
+
+    test "sends one WINDOW_UPDATE topping both windows back to peak once the threshold is crossed",
+         %{conn: conn} do
+      {conn, _ref} = open_request(conn)
+      assert_recv_frames [headers(stream_id: stream_id)]
+
+      # 60_000 bytes consumed drops both windows to exactly the 40_000
+      # threshold; the 7th frame lands after the refill so there's
+      # still just one WINDOW_UPDATE per window.
+      chunk = String.duplicate("a", 10_000)
+
+      frames = for _ <- 1..7, do: data(stream_id: stream_id, data: chunk)
+
+      assert {:ok, %HTTP2{} = _conn, _responses} = stream_frames(conn, frames)
+
+      assert_recv_frames [
+        window_update(stream_id: 0, window_size_increment: 60_000),
+        window_update(stream_id: ^stream_id, window_size_increment: 60_000)
+      ]
+    end
+
+    test "set_window_size/3 raises the target so subsequent refills top up to the new peak",
+         %{conn: conn} do
+      {conn, ref} = open_request(conn)
+      assert_recv_frames [headers(stream_id: stream_id)]
+
+      # Raise the connection peak mid-flight. set_window_size sends its own
+      # WINDOW_UPDATE for the bump; drain that before proceeding.
+      {:ok, conn} = HTTP2.set_window_size(conn, :connection, 500_000)
+      assert_recv_frames [window_update(stream_id: 0, window_size_increment: 400_000)]
+
+      # Also raise the stream peak so the stream doesn't bottleneck the
+      # connection-level test.
+      {:ok, conn} = HTTP2.set_window_size(conn, {:request, ref}, 500_000)
+      assert_recv_frames [window_update(stream_id: ^stream_id, window_size_increment: 400_000)]
+
+      # Consume enough to drop both windows to the 40_000 threshold;
+      # the refill should top up to the raised 500_000 peak, not the
+      # original 100_000 configured at connect.
+      chunk = String.duplicate("a", 10_000)
+      frames = for _ <- 1..46, do: data(stream_id: stream_id, data: chunk)
+
+      assert {:ok, %HTTP2{} = _conn, _responses} = stream_frames(conn, frames)
+
+      assert_recv_frames [
+        window_update(stream_id: 0, window_size_increment: 460_000),
+        window_update(stream_id: ^stream_id, window_size_increment: 460_000)
+      ]
+    end
   end
 
   describe "settings" do