fix(gateway): replay voice state updates after a resume (#2249)

fluxer_gateway/src/session/session_dispatch.erl

@@ -195,14 +195,45 @@ finalize_dispatch(Event, Data, State) ->
 do_handle_dispatch_pre_encoded(Event, {pre_encoded, EncodedData} = Data, State) ->
     Seq = maps:get(seq, State),
     NewSeq = Seq + 1,
+    BufferedState = buffer_pre_encoded_event(Event, Data, NewSeq, State),
     send_to_socket(maps:get(socket_pid, State, undefined), Event, Data, NewSeq),
     StateAfterMain =
         case needs_state_update(Event) of
-            true -> apply_pre_encoded_state_update(Event, EncodedData, State, NewSeq);
-            false -> State#{seq => NewSeq}
+            true -> apply_pre_encoded_state_update(Event, EncodedData, BufferedState, NewSeq);
+            false -> BufferedState#{seq => NewSeq}
         end,
     {noreply, StateAfterMain}.
 
+-spec buffer_pre_encoded_event(
+    event(), {pre_encoded, binary()}, non_neg_integer(), session_state()
+) ->
+    session_state().
+buffer_pre_encoded_event(Event, Data, NewSeq, State) ->
+    case should_buffer_pre_encoded(Event) of
+        false ->
+            State;
+        true ->
+            Request = #{event => Event, data => Data, seq => NewSeq},
+            RequestBytes = buffer_entry_bytes(Request),
+            case is_oversized_event(RequestBytes) of
+                true ->
+                    State;
+                false ->
+                    Buffer = maps:get(buffer, State),
+                    Deque =
+                        case is_list(Buffer) of
+                            true ->
+                                limited_deque:from_list(
+                                    Buffer, ?MAX_EVENT_BUFFER_SIZE, ?MAX_TOTAL_BUFFER_BYTES
+                                );
+                            false ->
+                                Buffer
+                        end,
+                    NewBuffer = limited_deque:push(Request, RequestBytes, Deque),
+                    State#{buffer => NewBuffer, buffer_bytes => limited_deque:bytes(NewBuffer)}
+            end
+    end.
+
 -spec apply_pre_encoded_state_update(event(), binary(), session_state(), non_neg_integer()) ->
     session_state().
 apply_pre_encoded_state_update(Event, EncodedData, State, NewSeq) ->
@@ -232,6 +263,10 @@ needs_state_update(_) -> false.
 is_oversized_event(RequestBytes) ->
     RequestBytes > ?MAX_SINGLE_EVENT_BUFFER_BYTES.
 
+-spec should_buffer_pre_encoded(event()) -> boolean().
+should_buffer_pre_encoded(Event) ->
+    event_name(Event) =:= <<"VOICE_STATE_UPDATE">>.
+
 -spec should_skip_replay_buffer(event()) -> boolean().
 should_skip_replay_buffer(Event) ->
     event_name(Event) =:= <<"GUILD_MEMBERS_CHUNK">>.

fluxer_gateway/test/session_dispatch_integration_tests.erl

@@ -138,6 +138,24 @@ pre_encoded_not_buffered_test() ->
     ?assertEqual(0, limited_deque:size(maps:get(buffer, State1))),
     ?assertEqual(1, maps:get(seq, State1)).
 
+pre_encoded_voice_state_is_buffered_for_replay_test() ->
+    State0 = base_state(#{}),
+    Data = {pre_encoded, <<"{\"user_id\":\"1\",\"channel_id\":\"2\"}">>},
+    {noreply, State1} = session_dispatch:handle_dispatch(voice_state_update, Data, State0),
+    ?assertEqual(1, limited_deque:size(maps:get(buffer, State1))),
+    [Entry] = limited_deque:to_list(maps:get(buffer, State1)),
+    ?assertEqual(voice_state_update, maps:get(event, Entry)),
+    ?assertEqual(Data, maps:get(data, Entry)),
+    ?assertEqual(1, maps:get(seq, Entry)).
+
+pre_encoded_member_list_stays_out_of_replay_test() ->
+    State0 = base_state(#{}),
+    Data = {pre_encoded, <<"[{\"test\":true}]">>},
+    {noreply, State1} = session_dispatch:handle_dispatch(
+        guild_member_list_update, Data, State0
+    ),
+    ?assertEqual(0, limited_deque:size(maps:get(buffer, State1))).
+
 pre_encoded_increments_seq_test() ->
     State0 = base_state(#{seq => 10}),
     Data = {pre_encoded, <<"[{\"test\":true}]">>},