Fail requests pipelined behind a 101 response

lib/mint/http1.ex

@@ -56,6 +56,11 @@ defmodule Mint.HTTP1 do
       `Mint.TransportError` with reason `:closed` instead, since the server might have
       processed them.
 
+    * `:connection_upgraded` - when a pipelined request gets no response because the
+      server answered a previous request with a `101` response. The connection then
+      uses another protocol, and the server might have received the request as part
+      of it.
+
     * `{:unexpected_data, data}` - when unexpected data is received from the server.
 
     * `:invalid_status_line` - when the HTTP/1 status line is invalid.
@@ -894,9 +899,12 @@ defmodule Mint.HTTP1 do
     decode(:status, conn, data, responses)
   end
 
+  # A 101 response switches the connection to another protocol (RFC 9110
+  # 15.2.2), so the bytes after it belong to that protocol.
   defp decode_body(:single, conn, data, _request_ref, responses) do
     {conn, responses} = add_body(conn, data, responses)
     {conn, responses} = request_done(conn, responses)
+    {conn, responses} = fail_requests(conn, responses, wrap_error(:connection_upgraded))
     {:ok, conn, responses}
   end
 
@@ -1205,14 +1213,28 @@ defmodule Mint.HTTP1 do
   # Requests pipelined behind a response that closes the connection never get a
   # response of their own.
   defp close_after_response(conn, responses, error) do
+    {conn, responses} = fail_requests(conn, responses, error)
+    {internal_close(conn), responses}
+  end
+
+  # Returns an error response with the given error for the current request and
+  # every queued one, and removes them from the connection. If one of them is
+  # streaming its body, the streaming state is cleared so the rest of the body
+  # can't be sent.
+  defp fail_requests(conn, responses, error) do
     requests = if conn.request, do: [conn.request | :queue.to_list(conn.requests)], else: []
 
     responses =
       Enum.reduce(requests, responses, fn request, responses ->
         [{:error, request.ref, error} | responses]
       end)
 
-    {internal_close(%{conn | request: nil, requests: :queue.new()}), responses}
+    # No request can be made while one is streaming its body, so a streaming
+    # request is the last one and it's among the failed ones unless it's done.
+    streaming_request = if requests == [], do: conn.streaming_request
+
+    conn = %{conn | request: nil, requests: :queue.new(), streaming_request: streaming_request}
+    {conn, responses}
   end
 
   defp pop_request(conn) do
@@ -1433,6 +1455,11 @@ defmodule Mint.HTTP1 do
       "\"connection: close\", so it's safe to retry on a new connection"
   end
 
+  def format_error(:connection_upgraded) do
+    "request got no response because the server switched the connection to another " <>
+      "protocol in response to a previous request"
+  end
+
   def format_error(:request_body_is_streaming) do
     "a request body is currently streaming, so no new requests can be issued"
   end

test/mint/http1/conn_test.exs

@@ -1240,6 +1240,71 @@
     assert conn.buffer == <<>>
   end
 
+  test "a request pipelined behind a successful CONNECT gets its response from the tunnel",
+       %{conn: conn} do
+    {:ok, conn, ref1} = HTTP1.request(conn, "CONNECT", "example.com:80", [], nil)
+    {:ok, conn, ref2} = HTTP1.request(conn, "GET", "/", [], nil)
+
+    assert {:ok, conn, [{:status, ^ref1, 200}, {:headers, ^ref1, []}, {:done, ^ref1}]} =
+             HTTP1.stream(
+               conn,
+               {:tcp, conn.socket, "HTTP/1.1 200 Connection Established\r\n\r\n"}
+             )
+
+    response = "HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok"
+
+    assert {:ok, conn,
+            [{:status, ^ref2, 200}, {:headers, ^ref2, _}, {:data, ^ref2, "ok"}, {:done, ^ref2}]} =
+             HTTP1.stream(conn, {:tcp, conn.socket, response})
+
+    assert HTTP1.open?(conn)
+  end
+
+  test "requests pipelined behind a 101 response get no response", %{conn: conn} do
+    {:ok, conn, ref1} = HTTP1.request(conn, "GET", "/", [], nil)
+    {:ok, conn, ref2} = HTTP1.request(conn, "GET", "/", [], nil)
+
+    response = "HTTP/1.1 101 Switching Protocols\r\nupgrade: foo\r\nconnection: upgrade\r\n\r\n"
+
+    assert {:ok, conn, responses} = HTTP1.stream(conn, {:tcp, conn.socket, response})
+
+    assert [
+             {:status, ^ref1, 101},
+             {:headers, ^ref1, _},
+             {:done, ^ref1},
+             {:error, ^ref2, %HTTPError{reason: :connection_upgraded}}
+           ] = responses
+
+    assert HTTP1.open?(conn)
+    assert HTTP1.open_request_count(conn) == 0
+
+    data = "HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\nok"
+
+    assert {:error, _conn, %HTTPError{reason: {:unexpected_data, ^data}}, []} =
+             HTTP1.stream(conn, {:tcp, conn.socket, data})
+  end
+
+  test "a streaming request pipelined behind a 101 response can't stream its body",
+       %{conn: conn} do
+    {:ok, conn, ref1} = HTTP1.request(conn, "GET", "/", [], nil)
+    {:ok, conn, ref2} = HTTP1.request(conn, "POST", "/", [], :stream)
+
+    response = "HTTP/1.1 101 Switching Protocols\r\nupgrade: foo\r\nconnection: upgrade\r\n\r\n"
+
+    assert {:ok, conn,
+            [
+              {:status, ^ref1, 101},
+              {:headers, ^ref1, _},
+              {:done, ^ref1},
+              {:error, ^ref2, %HTTPError{reason: :connection_upgraded}}
+            ]} = HTTP1.stream(conn, {:tcp, conn.socket, response})
+
+    assert {:error, _conn, %HTTPError{reason: :unknown_request_to_stream}} =
+             HTTP1.stream_request_body(conn, ref2, "hello")
+
+    assert_raise ArgumentError, fn -> HTTP1.request_body_window(conn, ref2) end
+  end
+
   test "unallowed trailer headers are removed from the trailer headers", %{conn: conn} do
     {:ok, conn, ref} = HTTP1.request(conn, "GET", "/", [], nil)
 
@@ -2108,7 +2173,7 @@
                \
                """)
 
       {:ok, conn} = HTTP1.stream_request_body(conn, ref, "hello")
       assert receive_request_string(server_socket) == "5\r\nhello\r\n"
     end
 
@@ -2127,7 +2192,7 @@
                \
                """)
 
       {:ok, conn} = HTTP1.stream_request_body(conn, ref, "hello")
       assert receive_request_string(server_socket) == "5\r\nhello\r\n"
     end
 
@@ -2147,7 +2212,7 @@
                \
                """)
 
       {:ok, conn} = HTTP1.stream_request_body(conn, ref, "hello")
       assert receive_request_string(server_socket) == "5\r\nhello\r\n"
     end
 
@@ -2167,7 +2232,7 @@
                \
                """)
 
       {:ok, conn} = HTTP1.stream_request_body(conn, ref, "hello")
       assert receive_request_string(server_socket) == "hello"
     end