Files
fluxer/fluxer_gateway/test/session_connection_guild_tests.erl

564 lines
19 KiB
Erlang

%% SPDX-License-Identifier: AGPL-3.0-or-later
-module(session_connection_guild_tests).
-typing([eqwalizer]).
-include_lib("eunit/include/eunit.hrl").
do_guild_connect_skips_session_connect_when_cached_unavailable_test() ->
GuildId = 2002,
Attempt = 3,
Parent = self(),
TestRef = make_ref(),
GuildPid = spawn(fun() -> guild_stub_loop(Parent, TestRef) end),
ManagerPid = spawn(fun() -> manager_stub_loop(GuildId, GuildPid) end),
true = register(guild_manager, ManagerPid),
Features = [<<"UNAVAILABLE_FOR_EVERYONE">>],
GuildData = #{<<"guild">> => #{<<"features">> => Features}},
CacheState = #{id => GuildId, data => GuildData},
_ = guild_availability:update_unavailability_cache_for_state(CacheState),
try
ok = session_connection_guild:do_guild_connect(
connect_context(Parent, GuildId, Attempt)
),
{ok, Resp} = await_unavailable_result(GuildId, Attempt),
?assertEqual(true, maps:get(<<"unavailable">>, Resp)),
?assertNot(saw_stub_call(TestRef, 200))
after
safe_unregister(guild_manager, ManagerPid),
ManagerPid ! stop,
GuildPid ! stop,
CleanupState = #{id => GuildId, data => #{<<"guild">> => #{<<"features">> => []}}},
guild_availability:update_unavailability_cache_for_state(CleanupState)
end.
do_guild_connect_uses_session_connect_async_cast_test() ->
GuildId = 2004,
Parent = self(),
TestRef = make_ref(),
GuildPid = spawn(fun() -> guild_stub_loop(Parent, TestRef) end),
ManagerPid = spawn(fun() -> manager_stub_loop(GuildId, GuildPid) end),
true = register(guild_manager, ManagerPid),
SessionPid = spawn(fun capture_loop/0),
try
ok = session_connection_guild:do_guild_connect(connect_context(SessionPid, GuildId, 0)),
?assertMatch(
{session_connect_async, _},
await_stub_cast(TestRef, 1000)
)
after
SessionPid ! stop,
safe_unregister(guild_manager, ManagerPid),
ManagerPid ! stop,
GuildPid ! stop
end.
do_guild_connect_start_or_lookup_after_lookup_miss_test() ->
GuildId = 2005,
Parent = self(),
TestRef = make_ref(),
GuildPid = spawn(fun() -> guild_stub_loop(Parent, TestRef) end),
ManagerPid = spawn(fun() -> manager_lookup_miss_stub_loop(GuildId, GuildPid) end),
true = register(guild_manager, ManagerPid),
SessionPid = spawn(fun capture_loop/0),
try
ok = session_connection_guild:do_guild_connect(connect_context(SessionPid, GuildId, 0)),
?assertMatch(
{session_connect_async, _},
await_stub_cast(TestRef, 1000)
)
after
SessionPid ! stop,
safe_unregister(guild_manager, ManagerPid),
ManagerPid ! stop,
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, TestRef),
{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, TestRef),
?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,
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, Token, State0
),
[E] = maps:get(collected_guild_states, State1, []),
?assertEqual(true, maps:get(<<"unavailable">>, E)),
?assertEqual(unavailable, maps:get(GuildId, maps:get(guilds, State1))),
?assertNot(maps:is_key(GuildId, maps:get(guild_connect_inflight, State1, #{}))).
session_connect_failed_exhaustion_marks_unavailable_test() ->
GuildId = 9002,
Attempt = 25,
State0 = (exhausted_connect_state(<<"st2">>, GuildId, undefined))#{
presence_pid => undefined
},
{noreply, State1} = session_connection_guild:handle_result_internal(
GuildId,
Attempt,
{error, {session_connect_failed, worker_crashed}},
State0
),
[E] = maps:get(collected_guild_states, State1, []),
?assertEqual(true, maps:get(<<"unavailable">>, E)),
?assertEqual(unavailable, maps:get(GuildId, maps:get(guilds, State1))).
guild_connect_worker_down_exhaustion_clears_inflight_and_marks_unavailable_test() ->
GuildId = 9003,
Attempt = 25,
WorkerRef = make_ref(),
State0 = (exhausted_connect_state(<<"st3">>, GuildId, Attempt))#{
guild_connect_workers => #{WorkerRef => {GuildId, Attempt, self()}},
presence_pid => undefined
},
{guild_connect_worker, {noreply, State1}} =
session_connection_guild:handle_guild_connect_worker_down(WorkerRef, killed, State0),
?assertEqual(#{}, maps:get(guild_connect_inflight, State1)),
?assertEqual(#{}, maps:get(guild_connect_workers, State1)),
?assertEqual(unavailable, maps:get(GuildId, maps:get(guilds, State1))).
guild_connect_worker_normal_down_only_cleans_worker_monitor_test() ->
GuildId = 9004,
Attempt = 3,
WorkerRef = make_ref(),
State0 = (exhausted_connect_state(<<"st4">>, GuildId, Attempt))#{
guild_connect_workers => #{WorkerRef => {GuildId, Attempt, self()}}
},
{guild_connect_worker, {noreply, State1}} =
session_connection_guild:handle_guild_connect_worker_down(WorkerRef, normal, State0),
?assertEqual(#{GuildId => Attempt}, maps:get(guild_connect_inflight, State1)),
?assertEqual(#{}, maps:get(guild_connect_workers, State1)).
finalize_guild_connection_propagates_ready_stop_test() ->
GuildId = 4242,
GuildPid = spawn(fun wait_for_stop/0),
try
{stop, normal, S} = session_connection_guild:finalize_guild_connection(
GuildId,
GuildPid,
#{guilds => #{}},
fun(St) -> {stop, normal, St#{stopped => true}} end
),
?assert(maps:get(stopped, S))
after
GuildPid ! stop
end.
repair_stalled_guild_connects_requeues_unresolved_guilds_test() ->
State0 = #{
id => <<"repair1">>,
user_id => 100,
guilds => #{
1 => undefined,
2 => unavailable,
3 => cached_unavailable,
4 => {self(), make_ref()}
},
guild_connect_inflight => #{2 => 1}
},
State1 = session_connection_guild:repair_stalled_guild_connects(State0),
?assert(is_integer(maps:get(guild_connect_last_repair_at, State1))),
?assertEqual([{guild_connect, 1, 0}], collect_guild_connect_messages([])).
repair_stalled_guild_connects_throttles_repair_test() ->
Now = erlang:system_time(millisecond),
State0 = #{
id => <<"repair2">>,
user_id => 100,
guilds => #{1 => undefined},
guild_connect_last_repair_at => Now
},
State1 = session_connection_guild:repair_stalled_guild_connects(State0),
?assertEqual(Now, maps:get(guild_connect_last_repair_at, State1)),
?assertEqual([], collect_guild_connect_messages([])).
repair_stalled_guild_connects_limits_batch_test() ->
Guilds = maps:from_list([{Id, undefined} || Id <- lists:seq(1, 12)]),
State0 = #{
id => <<"repair3">>,
user_id => 100,
guilds => Guilds,
guild_connect_last_repair_at => 0
},
_State1 = session_connection_guild:repair_stalled_guild_connects(State0),
Expected = [{guild_connect, Id, 0} || Id <- lists:seq(1, 8)],
?assertEqual(Expected, collect_guild_connect_messages([])).
not_member_below_cap_schedules_retry_test() ->
GuildId = 9201,
Attempt = 0,
State0 = #{
id => <<"nm1">>,
user_id => 100,
guilds => #{GuildId => undefined},
ready => undefined
},
{noreply, State1} = session_connection_guild:handle_result_internal(
GuildId, Attempt, {error, not_member}, State0
),
?assert(maps:is_key(GuildId, maps:get(guilds, State1))),
receive
{guild_connect, GuildId, 1} -> ok
after 5000 ->
?assert(false, not_member_retry_not_scheduled)
end.
not_member_exhaustion_removes_guild_without_unavailable_marker_test() ->
GuildId = 9202,
OtherGuildId = 5550,
Attempt = 3,
State0 = (base_dispatch_state())#{
guilds => #{
GuildId => {self(), make_ref()},
OtherGuildId => {self(), make_ref()}
},
guild_subscription_state => #{GuildId => #{}},
active_guilds => sets:from_list([GuildId]),
collected_guild_states => [],
ready => undefined
},
{noreply, State1} = session_connection_guild:handle_result_internal(
GuildId, Attempt, {error, not_member}, State0
),
?assertNot(maps:is_key(GuildId, maps:get(guilds, State1))),
?assertEqual([], maps:get(collected_guild_states, State1)),
?assert(maps:is_key(OtherGuildId, maps:get(guilds, State1))),
?assertNot(maps:is_key(GuildId, maps:get(guild_subscription_state, State1))),
?assertNot(sets:is_element(GuildId, maps:get(active_guilds, State1))),
?assertEqual([], collect_guild_connect_messages([])).
not_member_removal_unblocks_initial_ready_test() ->
GuildId = 9203,
Attempt = 3,
State0 = (base_dispatch_state())#{
guilds => #{GuildId => undefined},
collected_guild_states => [],
ready => #{<<"guilds">> => []},
presence_pid => self(),
socket_pid => undefined
},
?assertMatch(
{stop, normal, _},
session_connection_guild:handle_result_internal(
GuildId, Attempt, {error, not_member}, State0
)
).
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, TestRef) ->
{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))),
SessionId = maps:get(id, State),
?assertMatch({set_session_push_hold, SessionId, _}, await_stub_cast(TestRef, 2000)),
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">>,
user_id => 100,
seq => 0,
buffer => [],
buffer_bytes => 0,
socket_pid => undefined,
channels => #{},
relationships => #{},
suppress_presence_updates => false,
pending_presences => [],
presence_pid => undefined,
ignored_events => #{},
debounce_reactions => false,
reaction_buffer => [],
reaction_buffer_timer => undefined
}.
connect_context(SessionPid, GuildId, Attempt) ->
#{
session_pid => SessionPid,
guild_id => GuildId,
attempt => Attempt,
session_id => <<"sa1">>,
user_id => 77,
bot => false,
is_staff => false,
initial_guild_id => undefined,
peer_ip => undefined,
user_data => #{<<"flags">> => <<"0">>}
}.
exhausted_connect_state(SessionId, GuildId, Attempt) ->
#{
id => SessionId,
user_id => 100,
guilds => #{GuildId => undefined},
guild_connect_inflight => maybe_inflight(GuildId, Attempt),
collected_guild_states => [],
ready => #{<<"guilds">> => []}
}.
maybe_inflight(_GuildId, undefined) ->
#{};
maybe_inflight(GuildId, Attempt) ->
#{GuildId => Attempt}.
wait_for_stop() ->
receive
stop -> ok
after 30000 -> ok
end.
capture_loop() ->
receive
stop -> ok;
_ -> capture_loop()
after 30000 -> ok
end.
await_stub_cast(Ref, T) ->
receive
{guild_stub_cast, Ref, M} -> M;
_ -> await_stub_cast(Ref, T)
after T -> timeout
end.
await_unavailable_result(G, A) ->
receive
{guild_connect_result, G, A, {ok_cached_unavailable, R}} -> {ok, R};
_ -> await_unavailable_result(G, A)
after 1000 -> timeout
end.
saw_stub_call(Ref, T) ->
receive
{guild_stub_called, Ref, _} -> true;
_ -> saw_stub_call(Ref, T)
after T -> false
end.
safe_unregister(N, P) ->
case whereis(N) of
P -> unregister(N);
_ -> ok
end,
ok.
manager_stub_loop(G, P) ->
receive
stop ->
ok;
{'$gen_call', F, {with_timeout, Request, _Timeout}} ->
reply_manager_stub_request(G, P, F, Request),
manager_stub_loop(G, P);
{'$gen_call', F, Request} ->
reply_manager_stub_request(G, P, F, Request),
manager_stub_loop(G, P);
_ ->
manager_stub_loop(G, P)
after 30000 -> ok
end.
reply_manager_stub_request(G, P, F, {lookup, G}) ->
gen_server:reply(F, {ok, P});
reply_manager_stub_request(G, P, F, {start_or_lookup, G}) ->
gen_server:reply(F, {ok, P});
reply_manager_stub_request(_G, _P, F, _Request) ->
gen_server:reply(F, {error, unsupported}).
manager_lookup_miss_stub_loop(G, P) ->
receive
stop ->
ok;
{'$gen_call', F, {with_timeout, Request, _Timeout}} ->
reply_lookup_miss_stub_request(G, P, F, Request),
manager_lookup_miss_stub_loop(G, P);
{'$gen_call', F, Request} ->
reply_lookup_miss_stub_request(G, P, F, Request),
manager_lookup_miss_stub_loop(G, P);
_ ->
manager_lookup_miss_stub_loop(G, P)
after 30000 -> ok
end.
reply_lookup_miss_stub_request(G, _P, F, {lookup, G}) ->
gen_server:reply(F, {error, not_found});
reply_lookup_miss_stub_request(G, P, F, {start_or_lookup, G}) ->
gen_server:reply(F, {ok, P});
reply_lookup_miss_stub_request(_G, _P, F, _Request) ->
gen_server:reply(F, {error, unsupported}).
collect_guild_connect_messages(Acc) ->
receive
{guild_connect, _GuildId, _Attempt} = Msg ->
collect_guild_connect_messages([Msg | Acc])
after 0 ->
lists:reverse(Acc)
end.
guild_stub_loop(Parent, Ref) ->
receive
stop ->
ok;
{'$gen_cast', M} ->
Parent ! {guild_stub_cast, Ref, M},
guild_stub_loop(Parent, Ref);
{'$gen_call', F, R} ->
Parent ! {guild_stub_called, Ref, R},
gen_server:reply(F, {ok, #{}}),
guild_stub_loop(Parent, Ref);
_ ->
guild_stub_loop(Parent, Ref)
after 30000 -> ok
end.