Skip to content

Flush the request body guard cross-process-safe #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
samwar merged 1 commit into from
Jul 20, 2026
Merged

Flush the request body guard cross-process-safe #2

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 93 additions & 42 deletions src/webmachine_request.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,12 @@

-define(IDLE_TIMEOUT, infinity).

%% Finite timeout used when we're only draining leftover request bytes
%% (see maybe_flush_req_body/1). Kept intentionally short so that a
%% misbehaving or already-drained peer can never wedge the request
%% pipeline. Legitimate streaming reads still use ?IDLE_TIMEOUT.
-define(FLUSH_TIMEOUT, 5000).

-type t() :: #wm_reqstate{}.
-export_type([t/0]).

Expand Down Expand Up @@ -588,7 +594,10 @@ expects_continue(PassedState) ->
sent_continue() ->
get(webmachine_sent_continue) =:= true.

recv_stream_body(PassedState=#wm_reqstate{reqdata=RD}, MaxHunkSize) ->
recv_stream_body(PassedState, MaxHunkSize) ->
recv_stream_body(PassedState, MaxHunkSize, ?IDLE_TIMEOUT).

recv_stream_body(PassedState=#wm_reqstate{reqdata=RD}, MaxHunkSize, Timeout) ->
case expects_continue(PassedState) and (not sent_continue()) of
true ->
put(webmachine_sent_continue, true),
Expand All @@ -608,28 +617,29 @@ recv_stream_body(PassedState=#wm_reqstate{reqdata=RD}, MaxHunkSize) ->
{<<>>, done};
chunked ->
start_recv_chunked_body(PassedState#wm_reqstate.socket,
MaxHunkSize);
MaxHunkSize, Timeout);
Length ->
recv_unchunked_body(PassedState#wm_reqstate.socket,
MaxHunkSize, Length)
MaxHunkSize, Length, Timeout)
end.

recv_unchunked_body(Socket, MaxHunk, DataLeft) ->
recv_unchunked_body(Socket, MaxHunk, DataLeft, Timeout) ->
case MaxHunk >= DataLeft of
true ->
case mochiweb_socket:recv(Socket,DataLeft,?IDLE_TIMEOUT) of
{ok,Data1} ->
case mochiweb_socket:recv(Socket, DataLeft, Timeout) of
{ok, Data1} ->
record_stream_progress(done),
{Data1, done};
{error, Error} ->
throw({webmachine_recv_error, Error})
end;
false ->
case mochiweb_socket:recv(Socket,MaxHunk,?IDLE_TIMEOUT) of
{ok,Data2} ->
case mochiweb_socket:recv(Socket, MaxHunk, Timeout) of
{ok, Data2} ->
Next = fun() ->
recv_unchunked_body(Socket, MaxHunk,
DataLeft-MaxHunk)
DataLeft - MaxHunk,
Timeout)
end,
record_stream_progress(Next),
{Data2, Next};
Expand All @@ -638,33 +648,35 @@ recv_unchunked_body(Socket, MaxHunk, DataLeft) ->
end
end.

start_recv_chunked_body(Socket, MaxHunk) ->
case read_chunk_length(Socket, false) of
start_recv_chunked_body(Socket, MaxHunk, Timeout) ->
case read_chunk_length(Socket, false, Timeout) of
0 ->
record_stream_progress(done),
{<<>>, done};
ChunkLength ->
recv_chunked_body(Socket,MaxHunk, ChunkLength)
recv_chunked_body(Socket, MaxHunk, ChunkLength, Timeout)
end.
recv_chunked_body(Socket, MaxHunk, LeftInChunk) ->
recv_chunked_body(Socket, MaxHunk, LeftInChunk, Timeout) ->
case MaxHunk >= LeftInChunk of
true ->
case mochiweb_socket:recv(Socket,LeftInChunk,?IDLE_TIMEOUT) of
{ok,Data1} ->
case mochiweb_socket:recv(Socket, LeftInChunk, Timeout) of
{ok, Data1} ->
Next = fun() ->
start_recv_chunked_body(Socket, MaxHunk)
start_recv_chunked_body(Socket, MaxHunk,
Timeout)
end,
record_stream_progress(Next),
{Data1, Next};
{error, Error} ->
throw({webmachine_recv_error, Error})
end;
false ->
case mochiweb_socket:recv(Socket,MaxHunk,?IDLE_TIMEOUT) of
{ok,Data2} ->
case mochiweb_socket:recv(Socket, MaxHunk, Timeout) of
{ok, Data2} ->
Next = fun() ->
recv_chunked_body(Socket, MaxHunk,
LeftInChunk-MaxHunk)
LeftInChunk - MaxHunk,
Timeout)
end,
record_stream_progress(Next),
{Data2, Next};
Expand All @@ -673,9 +685,9 @@ recv_chunked_body(Socket, MaxHunk, LeftInChunk) ->
end
end.

read_chunk_length(Socket, MaybeLastChunk) ->
read_chunk_length(Socket, MaybeLastChunk, Timeout) ->
mochiweb_socket:setopts(Socket, [{packet, line}]),
case mochiweb_socket:recv(Socket, 0, ?IDLE_TIMEOUT) of
case mochiweb_socket:recv(Socket, 0, Timeout) of
{ok, Header} ->
mochiweb_socket:setopts(Socket, [{packet, raw}]),
Splitter = fun (C) ->
Expand All @@ -689,7 +701,7 @@ read_chunk_length(Socket, MaybeLastChunk) ->
%% allow [badly formed] last chunk header to be
%% empty instead of '0' explicitly
if MaybeLastChunk -> 0;
true -> read_chunk_length(Socket, true)
true -> read_chunk_length(Socket, true, Timeout)
end;
_ ->
erlang:list_to_integer(Hex, 16)
Expand All @@ -713,28 +725,34 @@ maybe_flush_req_body(Req) ->
put(mochiweb_request_recv, true),
true;
undefined ->
case expects_continue(Req) and (not sent_continue()) of
case get(mochiweb_request_recv) of
true ->
%% If the request expected continue, but we didn't
%% send it, then tell mochiweb to ignore what the
%% content length or transfer encoding says about
%% a body.
put(mochiweb_request_recv, true),
%% The request body was already consumed via
%% wrq:req_body/1 (eager read). mochiweb's
%% request_recv flag is set but our own
%% webmachine_stream_progress was never touched
%% because the eager path bypasses
%% recv_stream_body/2. There is nothing left on
%% the socket to drain; attempting to recv here
%% would block on an already-empty socket.
true;
false ->
MaxFlush = max_flush_bytes(),
case MaxFlush of
0 ->
%% this server has been configured to
%% close connections on clients who send
%% requests whose bodies are ignored
false;
_ ->
%% There might be a body sitting out there we
%% haven't read - give it a try.
ReadSize = erlang:min(65535, MaxFlush),
flush_req_body(
catch recv_stream_body(Req, ReadSize), MaxFlush)
_ ->
%% Cross-process-safe fallback: some
%% applications dispatch body reads to a
%% separate worker process, so the
%% mochiweb_request_recv pdict flag never lands
%% on the mochiweb loop process even though the
%% body was consumed. The bodyfetch field is
%% set to `standard` or `stream` inside
%% call({req_body, MaxRecvBody}, ReqState) when
%% the body is read, and travels with the
%% ReqState by value across process boundaries
%% (unlike the pdict). Consult it before
%% falling into the drain path.
case Req#wm_reqstate.bodyfetch of
standard -> true;
stream -> true;
_ -> maybe_flush_undefined_drain(Req)
end
end;
Next ->
Expand All @@ -749,6 +767,39 @@ maybe_flush_req_body(Req) ->
end
end.

%% Handles the "we have no idea whether a body is on the wire" branch
%% of maybe_flush_req_body/1. Only reached when
%% get(mochiweb_request_recv) is not `true` (i.e. no eager read has
%% claimed the body), so an actual attempt to recv is warranted. The
%% recv is bounded by ?FLUSH_TIMEOUT rather than ?IDLE_TIMEOUT so that
%% a bogus content-length or a peer that closed without shutting down
%% write cleanly cannot wedge the request pipeline.
maybe_flush_undefined_drain(Req) ->
case expects_continue(Req) and (not sent_continue()) of
true ->
%% If the request expected continue, but we didn't send
%% it, then tell mochiweb to ignore what the content
%% length or transfer encoding says about a body.
put(mochiweb_request_recv, true),
true;
false ->
MaxFlush = max_flush_bytes(),
case MaxFlush of
0 ->
%% this server has been configured to close
%% connections on clients who send requests whose
%% bodies are ignored
false;
_ ->
%% There might be a body sitting out there we
%% haven't read - give it a bounded try.
ReadSize = erlang:min(65535, MaxFlush),
flush_req_body(
catch recv_stream_body(Req, ReadSize, ?FLUSH_TIMEOUT),
MaxFlush)
end
end.

max_flush_bytes() ->
application:get_env(webmachine, max_flush_bytes, 67108864).

Expand Down
Loading