pepepper/fluxer
1
0
Fork 0
mirror of https://github.com/fluxerapp/fluxer synced 2026-10-08 03:32:27 +09:00
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): stop stale guild connect timers aborting connects (#2301)

Browse Source
This commit is contained in:
Hampus
2026-09-01 00:23:42 +02:00
committed by GitHub
parent 2c4d182d1f
commit 7806d2ac02
6 changed files with 223 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fluxer_gateway/src/session/session.erl
+4 -3
View File
@@ -65,6 +65,7 @@
pending_presences => pending_presence_buffer(),
guild_connect_inflight => #{guild_id() => non_neg_integer()},
guild_connect_workers => #{reference() => {guild_id(), non_neg_integer(), pid()}},
guild_connect_timers => #{guild_id() => {reference(), reference()}},
voice_queue => queue:queue(map()),
voice_queue_timer => reference() | undefined,
debounce_reactions => boolean(),
@@ -225,10 +226,10 @@ handle_info({guild_connect, GuildId, Attempt}, State) when
session_connection:handle_guild_connect(GuildId, Attempt, State);
handle_info({guild_connect_result, _, _, _} = Msg, State) ->
handle_info_guild_connect_result(Msg, State);
handle_info({guild_connect_timeout, GuildId, Attempt}, State) when
is_integer(GuildId), is_integer(Attempt), Attempt >= 0
handle_info({guild_connect_timeout, GuildId, Attempt, Token}, State) when
is_integer(GuildId), is_integer(Attempt), Attempt >= 0, is_reference(Token)
->
session_connection:handle_guild_connect_timeout(GuildId, Attempt, State);
session_connection:handle_guild_connect_timeout(GuildId, Attempt, Token, State);
handle_info({call_reconnect, ChannelId, Attempt}, State) when
is_integer(ChannelId), is_integer(Attempt), Attempt >= 0
->
fluxer_gateway/src/session/session_connection.erl
+4 -4
View File
@@ -8,7 +8,7 @@
repair_presence_connection/1,
handle_guild_connect/3,
handle_guild_connect_result/4,
handle_guild_connect_timeout/3,
handle_guild_connect_timeout/4,
handle_call_reconnect/3
]).
@@ -53,10 +53,10 @@ handle_guild_connect(GuildId, Attempt, State) ->
handle_guild_connect_result(GuildId, Attempt, Result, State) ->
session_connection_guild:handle_guild_connect_result(GuildId, Attempt, Result, State).
-spec handle_guild_connect_timeout(guild_id(), attempt(), session_state()) ->
-spec handle_guild_connect_timeout(guild_id(), attempt(), reference(), session_state()) ->
session_result().
handle_guild_connect_timeout(GuildId, Attempt, State) ->
session_connection_guild:handle_guild_connect_timeout(GuildId, Attempt, State).
handle_guild_connect_timeout(GuildId, Attempt, Token, State) ->
session_connection_guild:handle_guild_connect_timeout(GuildId, Attempt, Token, State).
-spec handle_call_reconnect(channel_id(), attempt(), session_state()) ->
{noreply, session_state()}.
fluxer_gateway/src/session/session_connection_guild.erl
+46 -23
View File
@@ -6,7 +6,7 @@
-export([
handle_guild_connect/3,
handle_guild_connect_result/4,
handle_guild_connect_timeout/3,
handle_guild_connect_timeout/4,
handle_guild_connect_worker_down/3,
maybe_spawn_guild_connect/5,
finalize_guild_connection/4,
@@ -18,6 +18,7 @@
-export_type([session_state/0, guild_id/0, attempt/0, guild_connect_result/0, session_result/0]).
-define(GUILD_CONNECT_MAX_INFLIGHT, 32).
-define(GUILD_CONNECT_TIMEOUT_MS, 120000).
-define(GUILD_CONNECT_REPAIR_INTERVAL_MS, 30000).
-define(GUILD_CONNECT_REPAIR_LIMIT, 8).
-define(NOT_MEMBER_MAX_RETRIES, 3).
@@ -108,9 +109,15 @@ spawn_guild_connect_worker(GuildId, Attempt, SessionId, UserId, Bot, IsStaff, In
do_guild_connect_with_release(Ctx)
end),
Workers = guild_connect_workers(State),
Timers = guild_connect_timers(State),
Token = make_ref(),
TimerRef = erlang:send_after(
?GUILD_CONNECT_TIMEOUT_MS, self(), {guild_connect_timeout, GuildId, Attempt, Token}
),
State1 = State#{
guild_connect_inflight => Inflight,
guild_connect_workers => Workers#{WorkerRef => {GuildId, Attempt, WorkerPid}}
guild_connect_workers => Workers#{WorkerRef => {GuildId, Attempt, WorkerPid}},
guild_connect_timers => Timers#{GuildId => {Token, TimerRef}}
},
{noreply, State1}.
@@ -163,29 +170,30 @@ handle_guild_connect_result(GuildId, Attempt, Result, State) ->
{noreply, State}
end.
-spec handle_guild_connect_timeout(guild_id(), attempt(), session_state()) ->
-spec handle_guild_connect_timeout(guild_id(), attempt(), reference(), session_state()) ->
session_result().
handle_guild_connect_timeout(GuildId, Attempt, State) ->
Inflight0 = guild_connect_inflight(State),
case maps:find(GuildId, Inflight0) of
{ok, Attempt} ->
UserId = maps:get(user_id, State),
logger:warning(
"guild_connect_timeout: guild_id=~p"
" user_id=~p attempt=~p",
[GuildId, UserId, Attempt]
),
State1 = remove_pending_guild_connect(GuildId, Attempt, State),
session_connection_retry:retry_or_fail(
GuildId,
Attempt,
State1,
fun session_ready:mark_guild_unavailable/2
);
_ ->
{noreply, State}
handle_guild_connect_timeout(GuildId, Attempt, Token, State) ->
case maps:get(GuildId, guild_connect_timers(State), undefined) of
{Token, _TimerRef} -> expire_guild_connect(GuildId, Attempt, State);
_ -> {noreply, State}
end.
-spec expire_guild_connect(guild_id(), attempt(), session_state()) -> session_result().
expire_guild_connect(GuildId, Attempt, State) ->
UserId = maps:get(user_id, State),
logger:warning(
"guild_connect_timeout: guild_id=~p"
" user_id=~p attempt=~p",
[GuildId, UserId, Attempt]
),
State1 = remove_pending_guild_connect(GuildId, Attempt, State),
session_connection_retry:retry_or_fail(
GuildId,
Attempt,
State1,
fun session_ready:mark_guild_unavailable/2
).
-spec handle_guild_connect_worker_down(reference(), term(), session_state()) ->
{guild_connect_worker, session_result()} | not_guild_connect_worker.
handle_guild_connect_worker_down(WorkerRef, Reason, State) ->
@@ -256,7 +264,22 @@ remove_pending_guild_connect(GuildId, Attempt, State) ->
-spec remove_guild_connect_inflight(guild_id(), session_state()) -> session_state().
remove_guild_connect_inflight(GuildId, State) ->
State#{guild_connect_inflight => maps:remove(GuildId, guild_connect_inflight(State))}.
State1 = cancel_guild_connect_timer(GuildId, State),
State1#{guild_connect_inflight => maps:remove(GuildId, guild_connect_inflight(State1))}.
-spec guild_connect_timers(session_state()) -> map().
guild_connect_timers(State) ->
maps:get(guild_connect_timers, State, #{}).
-spec cancel_guild_connect_timer(guild_id(), session_state()) -> session_state().
cancel_guild_connect_timer(GuildId, State) ->
case maps:take(GuildId, guild_connect_timers(State)) of
{{_Token, TimerRef}, Remaining} ->
_ = erlang:cancel_timer(TimerRef),
State#{guild_connect_timers => Remaining};
_ ->
State
end.
-spec guild_connect_workers(session_state()) -> map().
guild_connect_workers(State) ->
fluxer_gateway/src/session/session_connection_guild_resolve.erl
+6 -12
View File
@@ -7,7 +7,6 @@
-export_type([guild_id/0, attempt/0, connect_ctx/0]).
-define(GUILD_CONNECT_ASYNC_TIMEOUT_MS, 120000).
-define(GUILD_MANAGER_START_TIMEOUT_MS, 20000).
-define(GUILD_MANAGER_LOOKUP_FALLBACK_TIMEOUT_MS, 200).
-define(MAX_GUILD_OWNER_REDIRECTS, 3).
@@ -210,27 +209,22 @@ do_start_connect_async(GuildPid, Ctx) ->
{session_connect_async, #{
guild_id => GuildId, attempt => Attempt, request => Request
}},
send_connect_cast(GuildPid, CastMsg, GuildId, Attempt, SessionPid).
send_connect_cast(GuildPid, CastMsg, SessionPid).
-spec send_connect_cast(pid(), term(), integer(), non_neg_integer(), pid()) ->
pending | {error, term()}.
send_connect_cast(GuildPid, CastMsg, GuildId, Attempt, SessionPid) ->
-spec send_connect_cast(pid(), term(), pid()) -> pending | {error, term()}.
send_connect_cast(GuildPid, CastMsg, SessionPid) ->
CastStartedAt = gateway_timings:start(),
Result = shard_utils:safe_cast(GuildPid, CastMsg),
notify_remote_timing(
SessionPid,
guild,
node(GuildPid),
<<"session_connection_guild_resolve:send_connect_cast/5">>,
<<"session_connection_guild_resolve:send_connect_cast/3">>,
CastStartedAt
),
case Result of
ok ->
TimeoutMsg = {guild_connect_timeout, GuildId, Attempt},
_ = erlang:send_after(?GUILD_CONNECT_ASYNC_TIMEOUT_MS, SessionPid, TimeoutMsg),
pending;
{error, overloaded} ->
{error, {guild_manager_failed, {error, overloaded}}}
ok -> pending;
{error, overloaded} -> {error, {guild_manager_failed, {error, overloaded}}}
end.
-spec notify_remote_timing(connect_ctx() | pid(), term(), node(), binary(), integer()) -> ok.
fluxer_gateway/src/session/session_init.erl
+1
View File
@@ -233,6 +233,7 @@ extract_extra_fields(D, Ready) ->
pending_presences => [],
guild_connect_inflight => #{},
guild_connect_workers => #{},
guild_connect_timers => #{},
debounce_reactions => maps:get(debounce_reactions, D, false),
reaction_buffer => [],
reaction_buffer_timer => undefined
fluxer_gateway/test/session_connection_guild_tests.erl
+162 -2
View File
@@ -74,12 +74,56 @@ do_guild_connect_start_or_lookup_after_lookup_miss_test() ->
GuildPid ! stop
end.
stale_guild_connect_timeout_cannot_abort_a_later_connect_test() ->
GuildId = 9101,
TestRef = make_ref(),
{GuildPid, ManagerPid} = start_guild_stubs(GuildId, TestRef),
Tracer = start_connect_timeout_tracer(),
try
State0 = stale_timer_state(<<"stale-connect">>, GuildId, 4242),
{State1, FirstTimeout} = begin_guild_connect(GuildId, State0, TestRef, Tracer),
State2 = complete_guild_connect(GuildId, GuildPid, State1),
{State3, SecondTimeout} = begin_guild_connect(GuildId, State2, TestRef, Tracer),
{noreply, State4} = session:handle_info(FirstTimeout, State3),
?assertEqual(0, maps:get(GuildId, maps:get(guild_connect_inflight, State4), missing)),
?assertNot(awaits_guild_connect_retry(GuildId, 1500)),
?assertNotEqual(FirstTimeout, SecondTimeout),
{noreply, State5} = session:handle_info(SecondTimeout, State4),
?assertNot(maps:is_key(GuildId, maps:get(guild_connect_inflight, State5))),
?assert(awaits_guild_connect_retry(GuildId, 5000))
after
stop_connect_timeout_tracer(Tracer),
stop_guild_stubs(GuildPid, ManagerPid)
end.
guild_connect_success_cancels_the_connect_timeout_test() ->
GuildId = 9102,
TestRef = make_ref(),
{GuildPid, ManagerPid} = start_guild_stubs(GuildId, TestRef),
try
State0 = stale_timer_state(<<"cancel-connect">>, GuildId, 4243),
{noreply, State1} = session_connection_guild:maybe_spawn_guild_connect(
GuildId, 0, <<"cancel-connect">>, 4243, State0
),
?assertMatch({session_connect_async, _}, await_stub_cast(TestRef, 2000)),
{_Token, TimerRef} = maps:get(GuildId, maps:get(guild_connect_timers, State1)),
?assert(is_integer(erlang:read_timer(TimerRef))),
State2 = complete_guild_connect(GuildId, GuildPid, State1),
?assertEqual(#{}, maps:get(guild_connect_timers, State2)),
?assertEqual(false, erlang:read_timer(TimerRef))
after
stop_guild_stubs(GuildPid, ManagerPid)
end.
guild_connect_timeout_exhaustion_marks_unavailable_test() ->
GuildId = 9001,
Attempt = 25,
State0 = exhausted_connect_state(<<"st1">>, GuildId, Attempt),
Token = make_ref(),
State0 = (exhausted_connect_state(<<"st1">>, GuildId, Attempt))#{
guild_connect_timers => #{GuildId => {Token, make_ref()}}
},
{noreply, State1} = session_connection_guild:handle_guild_connect_timeout(
GuildId, Attempt, State0
GuildId, Attempt, Token, State0
),
[E] = maps:get(collected_guild_states, State1, []),
?assertEqual(true, maps:get(<<"unavailable">>, E)),
@@ -243,6 +287,122 @@ not_member_removal_unblocks_initial_ready_test() ->
)
).
start_guild_stubs(GuildId, TestRef) ->
Parent = self(),
GuildPid = spawn(fun() -> guild_stub_loop(Parent, TestRef) end),
ManagerPid = spawn(fun() -> manager_stub_loop(GuildId, GuildPid) end),
true = register(guild_manager, ManagerPid),
{GuildPid, ManagerPid}.
stop_guild_stubs(GuildPid, ManagerPid) ->
safe_unregister(guild_manager, ManagerPid),
ManagerPid ! stop,
GuildPid ! stop,
ok.
begin_guild_connect(GuildId, State, TestRef, Tracer) ->
{noreply, State1} = session_connection_guild:maybe_spawn_guild_connect(
GuildId, 0, maps:get(id, State), maps:get(user_id, State), State
),
?assertMatch({session_connect_async, _}, await_stub_cast(TestRef, 2000)),
{State1, await_connect_timeout(Tracer, 2000)}.
complete_guild_connect(GuildId, GuildPid, State) ->
{noreply, State1} = session_connection_guild:handle_guild_connect_result(
GuildId, 0, {ok, GuildPid, guild_state_payload(GuildId)}, State
),
?assertMatch({GuildPid, _}, maps:get(GuildId, maps:get(guilds, State1))),
dropped_guild_state(GuildId, State1).
stale_timer_state(SessionId, GuildId, UserId) ->
(base_dispatch_state())#{
id => SessionId,
user_id => UserId,
user_data => #{<<"flags">> => <<"0">>},
guilds => #{GuildId => undefined},
active_guilds => sets:new(),
guild_subscription_state => #{},
guild_connect_inflight => #{},
guild_connect_workers => #{},
collected_guild_states => [],
ready => undefined
}.
dropped_guild_state(GuildId, State) ->
Guilds = maps:get(guilds, State),
demonitor_guild_entry(maps:get(GuildId, Guilds, undefined)),
State#{guilds => Guilds#{GuildId => undefined}}.
demonitor_guild_entry({_Pid, Ref}) when is_reference(Ref) ->
demonitor(Ref, [flush]),
ok;
demonitor_guild_entry(_) ->
ok.
guild_state_payload(GuildId) ->
#{<<"id">> => integer_to_binary(GuildId), <<"name">> => <<"stale">>}.
awaits_guild_connect_retry(GuildId, Timeout) ->
receive
{guild_connect, GuildId, 1} -> true
after Timeout -> false
end.
start_connect_timeout_tracer() ->
Tracer = spawn(fun() -> connect_timeout_tracer_loop([]) end),
erlang:trace_pattern({erlang, send_after, 3}, true, [global]),
erlang:trace(all, true, [call, {tracer, Tracer}]),
Tracer.
stop_connect_timeout_tracer(Tracer) ->
erlang:trace(all, false, [call]),
erlang:trace_pattern({erlang, send_after, 3}, false, [global]),
Tracer ! stop,
ok.
connect_timeout_tracer_loop(Acc) ->
receive
stop ->
ok;
{take, From} ->
connect_timeout_tracer_loop(reply_traced_timeout(From, Acc));
{trace, _Pid, call, {erlang, send_after, [_Delay, _Dest, Msg]}} ->
connect_timeout_tracer_loop(Acc ++ traced_connect_timeout(Msg));
_ ->
connect_timeout_tracer_loop(Acc)
after 60000 -> ok
end.
traced_connect_timeout(Msg) when is_tuple(Msg), tuple_size(Msg) >= 3 ->
connect_timeout_only(element(1, Msg), Msg);
traced_connect_timeout(_Msg) ->
[].
connect_timeout_only(guild_connect_timeout, Msg) -> [Msg];
connect_timeout_only(_Tag, _Msg) -> [].
reply_traced_timeout(From, []) ->
From ! {traced_timeout, empty},
[];
reply_traced_timeout(From, [Msg | Rest]) ->
From ! {traced_timeout, Msg},
Rest.
await_connect_timeout(Tracer, Timeout) when Timeout =< 0 ->
Tracer,
?assert(false, connect_timeout_never_armed);
await_connect_timeout(Tracer, Timeout) ->
Tracer ! {take, self()},
receive
{traced_timeout, empty} ->
timer:sleep(20),
await_connect_timeout(Tracer, Timeout - 20);
{traced_timeout, Msg} ->
Msg
after 1000 ->
?assert(false, tracer_did_not_reply)
end.
base_dispatch_state() ->
#{
id => <<"nm">>,