diff --git a/grpc/lib/grpc/client/adapters/gun.ex b/grpc/lib/grpc/client/adapters/gun.ex index 7e53a9db7..8a8b4dd32 100644 --- a/grpc/lib/grpc/client/adapters/gun.ex +++ b/grpc/lib/grpc/client/adapters/gun.ex @@ -358,16 +358,17 @@ if Code.ensure_loaded?(:gun) do end defp recv_body(stream_payload, opts) do - recv_body(stream_payload, "", opts) + recv_body(stream_payload, [], opts) end defp recv_body(stream_payload, acc, opts) do case recv_data_or_trailers(stream_payload, opts) do {:data, data} -> - recv_body(stream_payload, <>, opts) + recv_body(stream_payload, [data | acc], opts) {:trailers, trailers} -> - {:ok, acc, GRPC.Transport.HTTP2.decode_headers(trailers)} + body = acc |> Enum.reverse() |> IO.iodata_to_binary() + {:ok, body, GRPC.Transport.HTTP2.decode_headers(trailers)} err -> err diff --git a/grpc_server/lib/grpc/server/adapters/cowboy/handler.ex b/grpc_server/lib/grpc/server/adapters/cowboy/handler.ex index c2a9fccb0..da509e7d0 100644 --- a/grpc_server/lib/grpc/server/adapters/cowboy/handler.ex +++ b/grpc_server/lib/grpc/server/adapters/cowboy/handler.ex @@ -303,7 +303,7 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do # APIs end def info({:read_full_body, ref, pid}, req, state) do - {s, body, req} = read_full_body(req, <<>>, state[:handling_timer], state.max_body_size) + {s, body, req} = read_full_body(req, [], 0, state[:handling_timer], state.max_body_size) send(pid, {ref, {s, body}}) {:ok, req, state} catch @@ -585,26 +585,26 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do end end - defp read_full_body(req, body, timer, max_bytes) do + defp read_full_body(req, chunks, size, timer, max_bytes) do result = :cowboy_req.read_body(req, timeout_left_opt(timer)) case result do {:ok, data, req} -> - total = body <> data + size = size + byte_size(data) - if byte_size(total) > max_bytes do - throw({:body_too_large, byte_size(total)}) + if size > max_bytes do + throw({:body_too_large, size}) else - {:ok, total, req} + {:ok, [data | chunks] |> Enum.reverse() |> IO.iodata_to_binary(), req} end {:more, data, req} -> - total = body <> data + size = size + byte_size(data) - if byte_size(total) > max_bytes do - throw({:body_too_large, byte_size(total)}) + if size > max_bytes do + throw({:body_too_large, size}) else - read_full_body(req, total, timer, max_bytes) + read_full_body(req, [data | chunks], size, timer, max_bytes) end end end diff --git a/grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs b/grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs index a7971478b..ded93b633 100644 --- a/grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs +++ b/grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs @@ -73,6 +73,39 @@ defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do end end + # Like collect_grpc_status/2, but also decodes the accumulated body as a HelloReply. + defp collect_grpc_response(conn, stream_ref) do + collect_grpc_response(conn, stream_ref, nil, <<>>) + end + + defp collect_grpc_response(conn, stream_ref, last_status, body) do + case :gun.await(conn, stream_ref, 5_000) do + {:response, :fin, _http_status, headers} -> + {find_grpc_status(headers) || last_status, decode_hello_reply(body)} + + {:response, :nofin, _http_status, headers} -> + collect_grpc_response(conn, stream_ref, find_grpc_status(headers), body) + + {:data, :fin, data} -> + {last_status, decode_hello_reply(body <> data)} + + {:data, :nofin, data} -> + collect_grpc_response(conn, stream_ref, last_status, body <> data) + + {:trailers, trailers} -> + {find_grpc_status(trailers) || last_status, decode_hello_reply(body)} + + {:error, reason} -> + flunk("gun error: #{inspect(reason)}") + end + end + + defp decode_hello_reply(<<_flag::8, length::32, message::bytes-size(length), _rest::binary>>) do + Protobuf.decode(message, Helloworld.HelloReply) + end + + defp decode_hello_reply(_incomplete), do: nil + # -------------------------------------------------------------------------- # Tests: max_body_size enforcement # -------------------------------------------------------------------------- @@ -155,6 +188,75 @@ defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do end end + # -------------------------------------------------------------------------- + # Tests: request body delivered across multiple HTTP/2 DATA frames + # -------------------------------------------------------------------------- + + describe "streamed request body" do + # read_full_body/5 recurses once per :more read; every other test in this + # file sends a body small enough to complete in a single :ok read, so + # this is the only test that exercises that recursive accumulation path. + test "reassembles a request body sent as several separate DATA frames, in order" do + run_server_with_opts([HelloServer], [], fn port -> + # Distinct segments so a dropped, duplicated, or reordered chunk changes the decoded name. + chunk_a = String.duplicate("a", 20_000) + chunk_b = String.duplicate("b", 20_000) + chunk_c = String.duplicate("c", 20_000) + name = chunk_a <> chunk_b <> chunk_c + + body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: name})) + <> = body + + conn = open_h2(port) + + start_tracing_read_full_body() + + stream_ref = + :gun.headers(conn, "POST", "/helloworld.Greeter/SayHello", grpc_request_headers()) + + # Waiting for read_full_body to recurse before each send proves the + # chunk just sent already landed in its own read_body call. + await_read_full_body_call() + :gun.data(conn, stream_ref, :nofin, part1) + await_read_full_body_call() + :gun.data(conn, stream_ref, :nofin, part2) + await_read_full_body_call() + :gun.data(conn, stream_ref, :fin, part3) + + assert {"0", reply} = collect_grpc_response(conn, stream_ref) + assert reply.message == "Hello, #{name}" + + stop_tracing_read_full_body() + :gun.close(conn) + end) + end + end + + # read_full_body/5 is private; :local makes trace_pattern instrument it anyway. + defp start_tracing_read_full_body do + Code.ensure_loaded!(GRPC.Server.Adapters.Cowboy.Handler) + + :erlang.trace_pattern({GRPC.Server.Adapters.Cowboy.Handler, :read_full_body, :_}, true, [ + :local + ]) + + :erlang.trace(:all, true, [:call]) + end + + defp stop_tracing_read_full_body do + :erlang.trace(:all, false, [:call]) + + :erlang.trace_pattern({GRPC.Server.Adapters.Cowboy.Handler, :read_full_body, :_}, false, [ + :local + ]) + end + + defp await_read_full_body_call do + assert_receive {:trace, _pid, :call, + {GRPC.Server.Adapters.Cowboy.Handler, :read_full_body, _args}}, + 2_000 + end + # -------------------------------------------------------------------------- # Private helper: start a server with specific opts and run a test function # --------------------------------------------------------------------------