Improve unrecoverable error handling (#433)

* Pull socket level errors into their own exception

* Properly surface socket level errors on Plug.Conn.chunk/2 calls

* Rename SocketError to TransportError

* Handle RST_STREAM messages when writing H2 responses

* Improve error output tests per recommendation

lib/bandit/adapter.ex

@@ -186,6 +186,7 @@ defmodule Bandit.Adapter do
         {:ok, nil, send_data(adapter, encoded_chunk, false)}
       end
     rescue
+      error in Bandit.TransportError -> {:error, error.error}
       error -> {:error, Exception.message(error)}
     end
   end

lib/bandit/http1/socket.ex

@@ -290,7 +290,7 @@ defmodule Bandit.HTTP1.Socket do
       case ThousandIsland.Socket.recv(socket, 0, read_timeout) do
         {:ok, chunk} -> chunk
         {:error, :timeout} -> <<>>
-        {:error, reason} -> request_error!(reason)
+        {:error, reason} -> socket_error!(reason)
       end
     end
 
@@ -318,7 +318,7 @@ defmodule Bandit.HTTP1.Socket do
           request_error!("Body read timeout", :request_timeout)
 
         {:error, reason} ->
-          request_error!(reason)
+          socket_error!(reason)
       end
     end
 
@@ -380,7 +380,7 @@ defmodule Bandit.HTTP1.Socket do
 
       case ThousandIsland.Socket.sendfile(socket.socket, path, offset, length) do
         {:ok, _bytes_written} -> %{socket | write_state: :sent}
-        {:error, reason} -> request_error!(reason)
+        {:error, reason} -> socket_error!(reason)
       end
     end
 
@@ -393,7 +393,7 @@ defmodule Bandit.HTTP1.Socket do
         {:error, reason} ->
           # Prevent error handlers from possibly trying to send again
           send(self(), {:plug_conn, :sent})
-          request_error!(reason)
+          socket_error!(reason)
       end
     end
 
@@ -408,6 +408,8 @@ defmodule Bandit.HTTP1.Socket do
 
     def supported_upgrade?(_socket, protocol), do: protocol == :websocket
 
+    def send_on_error(%@for{}, %Bandit.TransportError{}), do: :ok
+
     def send_on_error(%@for{} = socket, error) do
       receive do
         {:plug_conn, :sent} -> %{socket | write_state: :sent}
@@ -428,5 +430,10 @@ defmodule Bandit.HTTP1.Socket do
     defp request_error!(reason, plug_status \\ :bad_request) do
       raise Bandit.HTTPError, message: to_string(reason), plug_status: plug_status
     end
+
+    @spec socket_error!(term()) :: no_return()
+    defp socket_error!(reason) do
+      raise Bandit.TransportError, message: "Unrecoverable error: #{reason}", error: reason
+    end
   end
 end

lib/bandit/http2/errors.ex

@@ -36,11 +36,17 @@ defmodule Bandit.HTTP2.Errors do
     http_1_1_requires: 0xD
   }
 
+  @spec to_reason(integer()) :: atom()
+
   for {name, value} <- error_codes do
     @spec unquote(name)() :: unquote(Macro.var(name, Elixir)) :: unquote(value)
     def unquote(name)(), do: unquote(value)
+
+    def to_reason(unquote(value)), do: unquote(name)
   end
 
+  def to_reason(_), do: :unknown
+
   # Represents a stream error as defined in RFC9113§5.4.2
   defmodule StreamError do
     defexception [:message, :error_code]

lib/bandit/http2/stream.ex

@@ -380,8 +380,18 @@ defmodule Bandit.HTTP2.Stream do
     end
 
     @spec do_recv_rst_stream!(term(), term()) :: no_return()
-    defp do_recv_rst_stream!(_stream, error_code),
-      do: raise("Client sent RST_STREAM with error code #{error_code}")
+    defp do_recv_rst_stream!(_stream, error_code) do
+      case Bandit.HTTP2.Errors.to_reason(error_code) do
+        reason when reason in [:no_error, :cancel] ->
+          raise(Bandit.TransportError, message: "Client reset stream normally", error: :closed)
+
+        reason ->
+          raise(Bandit.TransportError,
+            message: "Received RST_STREAM from client: #{reason} (#{error_code})",
+            error: reason
+          )
+      end
+    end
 
     @spec do_stream_closed_error!(term()) :: no_return()
     defp do_stream_closed_error!(msg), do: stream_error!(msg, Bandit.HTTP2.Errors.stream_closed())
@@ -418,6 +428,7 @@ defmodule Bandit.HTTP2.Stream do
       stream =
         receive do
           {:send_window_update, delta} -> do_recv_send_window_update(stream, delta)
+          {:rst_stream, error_code} -> do_recv_rst_stream!(stream, error_code)
         after
           0 -> stream
         end

lib/bandit/pipeline.ex

@@ -210,6 +210,7 @@ defmodule Bandit.Pipeline do
   defp handle_error(:error, %type{} = error, stacktrace, transport, span, opts, _conn)
        when type in [
               Bandit.HTTPError,
+              Bandit.TransportError,
               Bandit.HTTP2.Errors.StreamError,
               Bandit.HTTP2.Errors.ConnectionError
             ] do
@@ -243,7 +244,7 @@ defmodule Bandit.Pipeline do
     end
   end
 
-  def maybe_log_error(%Bandit.HTTPError{message: "closed"} = error, stacktrace, http_opts) do
+  def maybe_log_error(%Bandit.TransportError{error: :closed} = error, stacktrace, http_opts) do
     do_maybe_log_error(error, stacktrace, Keyword.get(http_opts, :log_client_closures, false))
   end
 

lib/bandit/transport_error.ex

@@ -0,0 +1,6 @@
+defmodule Bandit.TransportError do
+  # Represents an error coming from the underlying transport which cannot be signalled back to the
+  # client by conventional means within the request. Examples include TCP socket closures and
+  # errors in the case of HTTP/1, and stream resets in HTTP/2
+  defexception message: nil, error: nil
+end

test/bandit/http1/request_test.exs

@@ -111,7 +111,7 @@ defmodule HTTP1RequestTest do
           Process.sleep(500)
         end)
 
-      assert output =~ "[error] ** (Bandit.HTTPError) closed"
+      assert output =~ "[error] ** (Bandit.TransportError) Unrecoverable error: closed"
 
       # Make sure we don't log a stacktrace
       refute output =~ "lib/bandit/pipeline.ex:"
@@ -130,7 +130,7 @@ defmodule HTTP1RequestTest do
           Process.sleep(500)
         end)
 
-      assert output =~ "[error] ** (Bandit.HTTPError) closed"
+      assert output =~ "[error] ** (Bandit.TransportError) Unrecoverable error: closed"
 
       # Make sure we log a stacktrace
       assert output =~ "lib/bandit/pipeline.ex:"
@@ -978,7 +978,7 @@ defmodule HTTP1RequestTest do
           Process.sleep(1100)
         end)
 
-      assert errors =~ "(Bandit.HTTPError) Body read timeout"
+      assert errors =~ "[error] ** (Bandit.HTTPError) Body read timeout"
     end
 
     def short_body(conn) do
@@ -1054,7 +1054,8 @@ defmodule HTTP1RequestTest do
           Process.sleep(100)
         end)
 
-      assert errors =~ "(ArgumentError) upgrade to unsupported not supported by Bandit.Adapter"
+      assert errors =~
+               "[error] ** (ArgumentError) upgrade to unsupported not supported by Bandit.Adapter"
     end
 
     def upgrade_unsupported(conn) do
@@ -1086,7 +1087,7 @@ defmodule HTTP1RequestTest do
           Process.sleep(100)
         end)
 
-      assert errors =~ "(Bandit.HTTPError) HTTP method POST unsupported"
+      assert errors =~ "[error] ** (Bandit.HTTPError) HTTP method POST unsupported"
     end
 
     test "returns a 400 and errors loudly in cases where an upgrade is indicated but upgrade header is incorrect",
@@ -1113,7 +1114,7 @@ defmodule HTTP1RequestTest do
         end)
 
       assert errors =~
-               "(Bandit.HTTPError) 'upgrade' header must contain 'websocket', got [\"NOPE\"]"
+               "[error] ** (Bandit.HTTPError) 'upgrade' header must contain 'websocket', got [\"NOPE\"]"
     end
 
     test "returns a 400 and errors loudly in cases where an upgrade is indicated but connection header is incorrect",
@@ -1140,7 +1141,7 @@ defmodule HTTP1RequestTest do
         end)
 
       assert errors =~
-               "(Bandit.HTTPError) 'connection' header must contain 'upgrade', got [\"NOPE\"]"
+               "[error] ** (Bandit.HTTPError) 'connection' header must contain 'upgrade', got [\"NOPE\"]"
     end
 
     test "returns a 400 and errors loudly in cases where an upgrade is indicated but key header is incorrect",
@@ -1165,7 +1166,7 @@ defmodule HTTP1RequestTest do
           Process.sleep(100)
         end)
 
-      assert errors =~ "(Bandit.HTTPError) 'sec-websocket-key' header is absent"
+      assert errors =~ "[error] ** (Bandit.HTTPError) 'sec-websocket-key' header is absent"
     end
 
     test "returns a 400 and errors loudly in cases where an upgrade is indicated but version header is incorrect",
@@ -1192,7 +1193,7 @@ defmodule HTTP1RequestTest do
         end)
 
       assert errors =~
-               "(Bandit.HTTPError) 'sec-websocket-version' header must equal '13', got [\"99\"]"
+               "[error] ** (Bandit.HTTPError) 'sec-websocket-version' header must equal '13', got [\"99\"]"
     end
 
     test "returns a 400 and errors loudly if websocket support is not enabled", context do
@@ -2398,7 +2399,7 @@ defmodule HTTP1RequestTest do
           Process.sleep(500)
         end)
 
-      assert output =~ "(Bandit.HTTPError) closed"
+      assert output =~ "[error] ** (Bandit.TransportError) Unrecoverable error: closed"
       refute output =~ "IMPOSSIBLE"
       assert ThousandIsland.connection_pids(context[:server_pid]) == {:ok, []}
     end
@@ -2417,7 +2418,7 @@ defmodule HTTP1RequestTest do
       Transport.close(client)
 
       output = capture_log(fn -> Process.sleep(500) end)
-      assert output =~ "(Bandit.HTTPError) closed"
+      assert output =~ "[error] ** (Bandit.TransportError) Unrecoverable error: closed"
       refute output =~ "IMPOSSIBLE"
       assert ThousandIsland.connection_pids(context[:server_pid]) == {:ok, []}
     end
@@ -2448,7 +2449,7 @@ defmodule HTTP1RequestTest do
       conn = send_chunked(conn, 200)
 
       Process.sleep(200)
-      assert chunk(conn, "IMPOSSIBLE") == {:error, "closed"}
+      assert chunk(conn, "IMPOSSIBLE") == {:error, :closed}
 
       conn
     end
@@ -2462,7 +2463,7 @@ defmodule HTTP1RequestTest do
       Transport.close(client)
 
       output = capture_log(fn -> Process.sleep(500) end)
-      assert output =~ "(Bandit.HTTPError) closed"
+      assert output =~ "[error] ** (Bandit.TransportError) Unrecoverable error: closed"
       refute output =~ "IMPOSSIBLE"
       assert ThousandIsland.connection_pids(context[:server_pid]) == {:ok, []}
     end

test/bandit/http2/protocol_test.exs

@@ -96,7 +96,7 @@ defmodule HTTP2ProtocolTest do
         end)
 
       assert errors =~
-               "(Bandit.HTTP2.Errors.ConnectionError) Received WINDOW_UPDATE in idle state"
+               "[error] ** (Bandit.HTTP2.Errors.ConnectionError) Received WINDOW_UPDATE in idle state"
     end
 
     test "it should shut down the stream gracefully and log when encountering a stream error",
@@ -114,7 +114,7 @@ defmodule HTTP2ProtocolTest do
         end)
 
       assert errors =~
-               "(Bandit.HTTP2.Errors.StreamError) Received trailers with pseudo headers"
+               "[error] ** (Bandit.HTTP2.Errors.StreamError) Received trailers with pseudo headers"
     end
 
     test "stream errors are short logged by default", context do
@@ -1952,7 +1952,8 @@ defmodule HTTP2ProtocolTest do
           Process.sleep(100)
         end)
 
-      assert errors =~ "Terminating stream in remote_closed state"
+      assert errors =~
+               "[error] ** (Bandit.HTTP2.Errors.StreamError) Terminating stream in remote_closed state"
     end
 
     def no_response_get(conn) do
@@ -1974,7 +1975,8 @@ defmodule HTTP2ProtocolTest do
           Process.sleep(100)
         end)
 
-      assert errors =~ "Terminating stream in open state"
+      assert errors =~
+               "[error] ** (Bandit.HTTP2.Errors.StreamError) Terminating stream in open state"
     end
 
     def no_response_post(conn) do
@@ -1997,7 +1999,7 @@ defmodule HTTP2ProtocolTest do
       assert output =~ "(Bandit.HTTP2.Errors.ConnectionError) Received RST_STREAM in idle state"
     end
 
-    test "raises an error on upon receipt of an RST_STREAM frame", context do
+    test "raises an error upon receipt of an RST_STREAM frame during reading", context do
       socket = SimpleH2Client.setup_connection(context)
 
       errors =
@@ -2007,12 +2009,67 @@ defmodule HTTP2ProtocolTest do
           Process.sleep(100)
         end)
 
-      assert errors =~ "Client sent RST_STREAM with error code 99"
+      assert errors =~
+               "[error] ** (Bandit.TransportError) Received RST_STREAM from client: unknown (99)"
     end
 
     def expect_reset(conn) do
       read_body(conn)
     end
+
+    test "raises an error upon receipt of an RST_STREAM frame during writing", context do
+      socket = SimpleH2Client.setup_connection(context)
+
+      errors =
+        capture_log(fn ->
+          SimpleH2Client.send_simple_headers(socket, 1, :get, "/write_after_delay", context.port)
+          SimpleH2Client.send_rst_stream(socket, 1, 99)
+          Process.sleep(200)
+        end)
+
+      assert errors =~
+               "[error] ** (Bandit.TransportError) Received RST_STREAM from client: unknown (99)"
+    end
+
+    def write_after_delay(conn) do
+      Process.sleep(100)
+      send_resp(conn, 200, "OK")
+    end
+
+    test "considers :no_error RST_STREAM frame as a normal closure during chunk writing",
+         context do
+      socket = SimpleH2Client.setup_connection(context)
+
+      errors =
+        capture_log(fn ->
+          SimpleH2Client.send_simple_headers(socket, 1, :get, "/expect_chunk_error", context.port)
+          SimpleH2Client.send_rst_stream(socket, 1, 0)
+          Process.sleep(200)
+        end)
+
+      assert errors == ""
+    end
+
+    test "considers :cancel RST_STREAM frame as a normal closure during chunk writing",
+         context do
+      socket = SimpleH2Client.setup_connection(context)
+
+      errors =
+        capture_log(fn ->
+          SimpleH2Client.send_simple_headers(socket, 1, :get, "/expect_chunk_error", context.port)
+          SimpleH2Client.send_rst_stream(socket, 1, 8)
+          Process.sleep(200)
+        end)
+
+      assert errors == ""
+    end
+
+    def expect_chunk_error(conn) do
+      conn = send_chunked(conn, 200)
+      Process.sleep(100)
+      {:error, :closed} = chunk(conn, "CHUNK")
+      conn
+    end
   end
 
   describe "SETTINGS frames" do