diff --git a/CHANGELOG.md b/CHANGELOG.md index bfb1c9b6..c8971010 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **A caller that half-closes its QUIC write side still gets the provider's reply (macula#145).** A `client_stream` + caller that sends its STREAM_END and then closes its write direction — the usual half-close, and what macula-go + does — had its session ended by the station link on the transport's FIN, so the provider's STREAM_REPLY was dropped + and the caller saw its stream end with no frame at all. The link now tells the session that the peer finished + sending instead of ending it, and a session ends on that only where it still awaited the peer's last frame. + ## [17.1.1] - 2026-10-27 ### Fixed diff --git a/src/client/macula_station_link.erl b/src/client/macula_station_link.erl index 13789af3..a8503a9f 100644 --- a/src/client/macula_station_link.erl +++ b/src/client/macula_station_link.erl @@ -2181,18 +2181,17 @@ quic_event({quic, send_failed, Stream, Reason}, #state{stream_bufs = Bufs} = S) when is_map_key(Stream, Bufs) -> {noreply, end_sessions_on_stream(Stream, {send_failed, Reason}, S)}; -%% A dedicated stream QUIC closed, or the peer finished, under open sessions: -%% nothing more can arrive on it, so they end as for a failed write. The -%% station relay closes a session's streams only once both sides have ended, -%% or when it gives the session up (its lifetime, a lost relay), and a -%% half-close is a STREAM_END frame, never a QUIC close. Before these clauses -%% the event fell to the catch-all, and a reader waited for ever. +%% A dedicated stream QUIC closed under open sessions (the station relay gave the session up: +%% its lifetime ran out, or the relay was lost, or the peer reset its send side): nothing more +%% can arrive on it and nothing can be written, so they end as for a failed write. quic_event({quic, stream_closed, Stream, Flags}, #state{stream_bufs = Bufs} = S) when is_map_key(Stream, Bufs) -> {noreply, end_sessions_on_stream(Stream, {stream_closed, Flags}, S)}; +%% The peer finished sending on a dedicated stream: its QUIC FIN, the transport's half-close, +%% which is no session's end (`peer_finished_on_stream/2'). quic_event({quic, peer_send_shutdown, Stream, _}, #state{stream_bufs = Bufs} = S) when is_map_key(Stream, Bufs) -> - {noreply, end_sessions_on_stream(Stream, peer_send_shutdown, S)}; + {noreply, peer_finished_on_stream(Stream, S)}; %% A stream that closes before its STREAM_OPEN completed carries no session. quic_event({quic, Closed, Stream, _}, #state{opening_bufs = Opening} = S) when (Closed =:= stream_closed orelse Closed =:= peer_send_shutdown), @@ -4604,6 +4603,25 @@ unknown_frame_on_stream(Stream, Unknown, #state{reset_stream = Reset} = S) -> _ = Reset(Stream, unknown_frame_type), end_sessions_on_stream(Stream, {malformed, Unknown}, S). +%% The peer finished sending on `Stream' — its QUIC FIN — so no more of its frames will ever +%% arrive on it, though this side may still write. Each session the stream carries hears that and +%% decides: one that has already had the peer's last frame (an end, an error, a reply) has +%% everything it was waiting for and goes on writing, so a caller that half-closes its write side +%% after its own STREAM_END still gets the provider's reply (macula#145; a reply set afterwards +%% was dropped because the session had already been aborted and its routing gone). One still +%% waiting for that frame will never get it, and ends as a lost link ends it. A stream carrying +%% no session has nothing left to say at all, and closes here. +peer_finished_on_stream(Stream, #state{client_streams = CS, server_streams = SS} = S) -> + Carried = lists:usort([{Sid, Pid} || {Sid, {Pid, _Mon, On}} <- maps:to_list(CS) ++ maps:to_list(SS), + On =:= Stream]), + peer_finished_on_stream(Stream, Carried, S). + +peer_finished_on_stream(Stream, [], S) -> + end_sessions_on_stream(Stream, peer_send_shutdown, S); +peer_finished_on_stream(_Stream, Carried, S) -> + _ = [Pid ! {stream_read_closed, Sid, peer_send_shutdown} || {Sid, Pid} <- Carried], + S. + %% End every session carried by `Stream' once a write on it failed, or once %% its bytes turned out not to be frames this link accepts: abort each %% session's process with `Reason', as a lost link does, and drop its diff --git a/src/macula_stream.erl b/src/macula_stream.erl index f086a8b1..61c4c293 100644 --- a/src/macula_stream.erl +++ b/src/macula_stream.erl @@ -645,6 +645,13 @@ handle_info({reopened, Ref, Result}, #state{reopening = {Ref, Queued}} = State) handle_info({stream_write_failed, Sid, Reason}, #state{id = Sid} = State) -> {noreply, transport_failed({error, {transport, Reason}}, State)}; +%% The peer finished sending on this stream's transport (its QUIC write side closed), so no more of its frames will +%% ever arrive. A session that has already had the peer's last frame has all it was waiting for and goes on writing: a +%% caller that half-closes after its own STREAM_END still gets the provider's reply (macula#145). One still waiting for +%% that frame will never get it, and ends as a lost link ends it. +handle_info({stream_read_closed, Sid, Reason}, #state{id = Sid} = State) -> + {noreply, read_closed(Reason, State)}; + handle_info(_Msg, State) -> {noreply, State}. @@ -995,6 +1002,18 @@ ended_with(Err, State) -> transport_failed(Err, #state{reply = Reply} = State) -> session_ended(Err, ended_with(Err, State#state{peer = undefined, reply = first_reply(Reply, Err)})). +%% @private The peer's write side closed on the transport. A session that has already had the peer's last frame -- an +%% end, an error, a reply -- is not waiting for anything more of it, and keeps the end its own frames give it. One that +%% is still waiting for that frame ends as a lost link ends it, with the reason's name only. +read_closed(_Reason, #state{ended = Ended} = State) when Ended =/= undefined -> + State; +read_closed(_Reason, #state{closed_recv = true} = State) -> + State; +read_closed(_Reason, #state{reply = Reply} = State) when Reply =/= undefined -> + State; +read_closed(Reason, State) -> + abort_session(<<"disconnected">>, macula_reason_name:text(Reason), State). + %% @private A frame of a type that belongs on the control stream has no place %% on a dedicated stream, in either profile. It is the connection peer's %% doing, so the connection is rejected with malformed_frame, and this diff --git a/test/macula_station_link_stream_serving_tests.erl b/test/macula_station_link_stream_serving_tests.erl index 97fd5982..0eac16e0 100644 --- a/test/macula_station_link_stream_serving_tests.erl +++ b/test/macula_station_link_stream_serving_tests.erl @@ -53,7 +53,8 @@ stream_serving_test_() -> fun a_served_handler_sees_the_verified_caller/0, fun a_caller_serves_more_stream_opens_than_its_quota/0, fun a_copy_of_a_live_stream_open_keeps_its_place/0, - fun a_served_handler_sees_the_caller_of_a_non_map_open/0]] + fun a_served_handler_sees_the_caller_of_a_non_map_open/0, + fun a_callers_quic_half_close_after_its_stream_end_keeps_the_reply/0]] ++ [{"a served session leaves no process behind: " ++ Name, {timeout, 15, {spawn, fun() -> process_flag(trap_exit, true), served_sessions_end(Handler, Ending) end}}} || {Name, Handler, Ending} <- [{"the handler closes and returns", fun close_and_return/2, none}, @@ -579,6 +580,37 @@ a_served_handler_sees_the_caller_of_a_non_map_open() -> receive {text_caller_seen, _, _} = Seen -> Seen after ?EVENT_MS -> none end), stop(Link). +%% A caller that closes its QUIC write side once its STREAM_END is written -- the usual half-close, and what macula-go's +%% caller does -- has sent the transport's FIN and nothing about the session: a handler that answers only after it has +%% read the end still gets its STREAM_REPLY out, and the caller reads it (macula#145: the reply was thrown away and the +%% caller saw its stream end with no frame at all). +a_callers_quic_half_close_after_its_stream_end_keeps_the_reply() -> + #{link := Link} = World = linked(), + Test = self(), + Procedure = <<"foo.answers_after_eof">>, + ok = macula_station_link:advertise_stream(Link, ?REALM, Procedure, client_stream, + fun(Stream, _Args) -> + Test ! {session_served, self()}, + read_to_the_end(Stream, []), + Test ! read_the_end, + receive answer_now -> ok end, + macula_stream:set_reply(Stream, <<"x1,x2">>), + macula_stream:close(Stream) + end), + Caller = key(), + #{quic := Quic, frame := Open} = opened_by_peer(World, Caller, Procedure, #{}, #{mode => client_stream}), + Handler = receive {session_served, Served} -> Served after ?EVENT_MS -> erlang:error(session_not_served) end, + on_stream(Link, Quic, caller_half_end(Caller, Open)), + ?assertEqual(read_the_end, receive read_the_end -> read_the_end after ?EVENT_MS -> not_read end), + %% The caller's FIN, once the end is written and before the handler answers. The link has + %% taken it in when it answers its own state. + Link ! {quic, peer_send_shutdown, Quic, undefined}, + _ = sys:get_state(Link), + Handler ! answer_now, + {ok, Reply} = written_within(Quic, ?EVENT_MS), + ?assertMatch({ok, #{frame_type := stream_reply, payload := <<"x1,x2">>}}, provider_fields(Open, Reply)), + stop(Link). + read_to_the_end(Stream, _Args) -> read_until_ended(macula_stream:recv(Stream, 5_000), Stream). read_until_ended({chunk, _Chunk}, Stream) -> read_until_ended(macula_stream:recv(Stream, 5_000), Stream); @@ -832,6 +864,11 @@ caller_frame(Spec, Caller, OpenFrame) -> caller_stream_end(Caller, OpenFrame) -> caller_frame(#{frame_type => stream_end, seq => 0, role => both}, Caller, OpenFrame). +%% The caller's STREAM_END of its own direction alone: what a caller that goes on to read the +%% provider's answer sends, and what a caller that then closes its QUIC write side has wrapped up. +caller_half_end(Caller, OpenFrame) -> + caller_frame(#{frame_type => stream_end, seq => 0, role => send}, Caller, OpenFrame). + %% The fields of a provider frame the link wrote, verified as the provider's first frame under the open it answers. provider_fields(OpenFrame, Written) -> fields(macula_frame:verify_provider_stream(Written, macula_frame:open_stream(verified(OpenFrame)), profile())). diff --git a/test/macula_station_link_stream_session_tests.erl b/test/macula_station_link_stream_session_tests.erl index 6b6aaa15..a8a01d06 100644 --- a/test/macula_station_link_stream_session_tests.erl +++ b/test/macula_station_link_stream_session_tests.erl @@ -35,7 +35,8 @@ stream_session_test_() -> fun a_disconnect_ends_open_sessions/0, fun a_disconnect_tells_open_sessions_its_reasons_name_only/0, fun a_closed_dedicated_stream_ends_its_session/0, - fun a_dedicated_stream_the_peer_finished_ends_its_session/0]]. + fun a_dedicated_stream_the_peer_finished_ends_a_session_still_awaiting_it/0, + fun a_session_that_had_the_peers_last_frame_writes_on_after_its_quic_fin/0]]. %%------------------------------------------------------------------ %% Cases @@ -221,9 +222,8 @@ a_disconnect_tells_open_sessions_its_reasons_name_only() -> ?assertEqual({error, {<<"disconnected">>, <<"disconnected">>}}, macula_stream:await_reply(StreamPid, ?EVENT_MS)). %% A dedicated stream QUIC closes under an open session (the station relay gave the session up: its lifetime ran out, -%% or the relay was lost) ends that session: a reader waiting on it gets a disconnected error instead of waiting for -%% ever, the link closes the stream and stays up. A clean end never gets here: the relay closes a stream only once -%% both sides have ended, and a half-close is a STREAM_END, not a QUIC close. +%% or the relay was lost, or a peer reset its send side) ends that session: a reader waiting on it gets a disconnected +%% error instead of waiting for ever, the link closes the stream and stays up. A clean peer FIN is not this event. a_closed_dedicated_stream_ends_its_session() -> #{link := Link} = World = linked(), #{stream := StreamPid, quic := Quic} = opened(World, macula_node_keys:key_id(key()), bidi), @@ -233,8 +233,9 @@ a_closed_dedicated_stream_ends_its_session() -> ?assert(is_process_alive(Link)), stop(Link). -%% Likewise a stream the peer finished (a QUIC FIN) while the session was open: nothing more can arrive on it. -a_dedicated_stream_the_peer_finished_ends_its_session() -> +%% A stream the peer finished (a QUIC FIN) while the session still awaited the peer's last frame: none will ever +%% arrive on it, so the reader is told instead of waiting for ever. +a_dedicated_stream_the_peer_finished_ends_a_session_still_awaiting_it() -> #{link := Link} = World = linked(), #{stream := StreamPid, quic := Quic} = opened(World, macula_node_keys:key_id(key()), bidi), Link ! {quic, peer_send_shutdown, Quic, undefined}, @@ -242,6 +243,23 @@ a_dedicated_stream_the_peer_finished_ends_its_session() -> ?assert(is_process_alive(Link)), stop(Link). +%% But a peer that half-closes its QUIC write side once its own last frame is in has said everything this session +%% waits for, and the session goes on to answer it: the caller writes, and the write reaches the stream (macula#145, +%% the mirror of the served session whose reply was thrown away). +a_session_that_had_the_peers_last_frame_writes_on_after_its_quic_fin() -> + #{link := Link} = World = linked(), + Provider = key(), + #{stream := StreamPid, quic := Quic, open := Open} = opened(World, macula_node_keys:key_id(Provider), bidi), + on_stream(Link, Quic, provider_frame(#{frame_type => stream_end, seq => 0, role => send}, Provider, Open)), + ?assertEqual(eof, macula_stream:recv(StreamPid, ?EVENT_MS)), + Link ! {quic, peer_send_shutdown, Quic, undefined}, + _ = sys:get_state(Link), + ?assertEqual(ok, macula_stream:send(StreamPid, <<"still here">>)), + {ok, Written} = written_within(Quic, ?EVENT_MS), + ?assertMatch({ok, #{frame_type := stream_data, seq := 0, body := <<"still here">>}, _}, + macula_frame:verify_caller_stream(Written, macula_frame:open_stream(Open), profile())), + stop(Link). + %%------------------------------------------------------------------ %% Helpers %%------------------------------------------------------------------