From ce067c47eea3b432d2f91d72ebe36be2911478a1 Mon Sep 17 00:00:00 2001 From: beamologist Date: Wed, 28 Oct 2026 01:32:15 +0100 Subject: [PATCH] fix: a caller's QUIC half-close keeps the provider's reply (Refs #145) A caller of a client_stream sends its STREAM_END and then closes its QUIC write direction -- the usual half-close, and what macula-go does. The station link read the transport's FIN (peer_send_shutdown) as the stream's death and ended every session carried on it, so the provider's STREAM_REPLY was never written: its handler was aborted first, and the routing the reply would have gone out on was already gone. The caller saw its stream end with no frame at all (found by the release A0 cross-SDK interop, DESIGN_E2E_SEALED_BY_DEFAULT section 8.1, program 3). A FIN says that no more of the peer's frames will arrive on that stream; it says nothing about a session that may still write. The link now tells each session the stream carries that the peer finished sending, and the session 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, and one still waiting for that frame ends as a lost link ends it rather than waiting for ever. Tests: a served handler that answers only after it has read the caller's end still writes its STREAM_REPLY once the caller's FIN is in, and a caller session that has had the provider's last frame still sends on its stream after the FIN. Both fail without the fix. Co-Authored-By: Claude Code --- CHANGELOG.md | 8 ++++ src/client/macula_station_link.erl | 32 +++++++++++---- src/macula_stream.erl | 19 +++++++++ ...cula_station_link_stream_serving_tests.erl | 39 ++++++++++++++++++- ...cula_station_link_stream_session_tests.erl | 30 +++++++++++--- 5 files changed, 114 insertions(+), 14 deletions(-) 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 %%------------------------------------------------------------------