fix(gateway): clamp a heartbeat ack to the session sequence (#2283)

fluxer_gateway/src/session/session_lifecycle.erl

@@ -280,13 +280,14 @@ handle_is_staff(State) ->
 handle_heartbeat_ack(Seq, #{ack_seq := AckSeq} = State) when Seq < AckSeq ->
     {reply, true, State};
 handle_heartbeat_ack(Seq, #{buffer := Buffer} = State) ->
-    NewBuffer = drop_acked_buffer(Seq, Buffer),
+    AckedSeq = min(Seq, maps:get(seq, State, Seq)),
+    NewBuffer = drop_acked_buffer(AckedSeq, Buffer),
     NewBytes =
         case is_list(NewBuffer) of
             true -> session_init:replay_buffer_bytes(NewBuffer);
             false -> limited_deque:bytes(NewBuffer)
         end,
-    NewState0 = State#{ack_seq => Seq, buffer => NewBuffer, buffer_bytes => NewBytes},
+    NewState0 = State#{ack_seq => AckedSeq, buffer => NewBuffer, buffer_bytes => NewBytes},
     NewState = session_connection_guild:repair_stalled_guild_connects(NewState0),
     {reply, true, NewState}.
 

fluxer_gateway/test/session_lifecycle_tests.erl

@@ -151,6 +151,18 @@ heartbeat_ack_tolerates_backwards_ack_without_trimming_test() ->
     ?assertEqual(5, maps:get(ack_seq, State1)),
     ?assertEqual([Event], maps:get(buffer, State1)).
 
+heartbeat_ack_clamps_ack_seq_to_current_seq_test() ->
+    State0 = resume_test_state(#{
+        seq => 5,
+        ack_seq => 0,
+        buffer => [#{seq => 5, event => message_create, data => #{}}]
+    }),
+    {reply, true, State1} = session_lifecycle:handle_heartbeat_ack(1000000, State0),
+    ?assertEqual(5, maps:get(ack_seq, State1)),
+    ?assertMatch(
+        {reply, {ok, [], 5}, _}, session_lifecycle:handle_resume(5, self(), State1)
+    ).
+
 handle_resume_clamps_truncated_gap_in_replay_buffer_test() ->
     State0 = resume_test_state(#{
         seq => 5,