fix: a caller's QUIC half-close keeps the provider's reply (Refs #145)

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

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

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

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())).

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,15 +233,33 @@ 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},
     ?assertMatch({error, {<<"disconnected">>, _}}, macula_stream:recv(StreamPid, ?EVENT_MS)),
     ?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
 %%------------------------------------------------------------------