perf(gateway): speed up reconnects and pin guilds to nodes (#3107)

fluxer_gateway/test/fluxer_gateway_config_tests.erl

@@ -187,6 +187,40 @@ rpc_concurrency_key_defaults_test() ->
     ?assertEqual(512, maps:get(gateway_nats_rpc_max_handlers, Config)),
     ?assertEqual(512, maps:get(gateway_http_rpc_max_concurrency, Config)).
 
+pinned_node_defaults_keep_release_behaviour_test() ->
+    Config = fluxer_gateway_config:load(),
+    ?assertEqual(true, maps:get(nats_rpc_enabled, Config)),
+    ?assertEqual([], maps:get(pinned_guild_ids, Config)),
+    ?assertEqual(undefined, maps:get(guild_pin_keeper_beam, Config)).
+
+pinned_node_env_test() ->
+    with_envs(
+        [
+            {"FLUXER_GATEWAY_NATS_RPC_ENABLED", "false"},
+            {"FLUXER_GATEWAY_PINNED_GUILD_IDS", "1100000000000000001, 42"},
+            {"FLUXER_GATEWAY_GUILD_PIN_KEEPER_BEAM", "/etc/fluxer/gw/gateway_node_router.beam"},
+            {"FLUXER_GATEWAY_GUILD_PIN_KEEPER_BEAM_MD5", "D5E42B1D6D85C4CDEE93AA0CCA18A420"}
+        ],
+        fun() ->
+            Config = fluxer_gateway_config:load(),
+            ?assertEqual(false, maps:get(nats_rpc_enabled, Config)),
+            ?assertEqual([42, 1100000000000000001], maps:get(pinned_guild_ids, Config)),
+            ?assertEqual(
+                "/etc/fluxer/gw/gateway_node_router.beam",
+                maps:get(guild_pin_keeper_beam, Config)
+            ),
+            ?assertEqual(
+                <<"D5E42B1D6D85C4CDEE93AA0CCA18A420">>,
+                maps:get(guild_pin_keeper_beam_md5, Config)
+            )
+        end
+    ).
+
+pinned_guild_ids_reject_non_snowflakes_test() ->
+    with_env("FLUXER_GATEWAY_PINNED_GUILD_IDS", "1100000000000000001,ab", fun() ->
+        ?assertError({invalid_pinned_guild_id, "ab"}, fluxer_gateway_config:load())
+    end).
+
 optional_string_test() ->
     ?assertEqual(undefined, fluxer_gateway_config:optional_string(undefined)),
     ?assertEqual("hello", fluxer_gateway_config:optional_string(<<"hello">>)),

fluxer_gateway/test/gateway_stall_monitor_tests.erl

@@ -0,0 +1,93 @@
+%% SPDX-License-Identifier: AGPL-3.0-or-later
+
+-module(gateway_stall_monitor_tests).
+
+-include_lib("eunit/include/eunit.hrl").
+
+stall_monitor_test_() ->
+    {foreach, fun setup/0, fun cleanup/1, [
+        fun installs_and_releases_system_monitor/1,
+        fun counts_forwarded_events_and_names_the_process/1,
+        fun leaves_a_foreign_system_monitor_alone/1,
+        fun records_late_timer_ticks/1
+    ]}.
+
+setup() ->
+    _ = erlang:system_monitor(undefined),
+    application:unset_env(fluxer_gateway, stall_monitor_enabled),
+    ok.
+
+cleanup(_) ->
+    case whereis(gateway_stall_monitor) of
+        undefined -> ok;
+        Pid -> gen_server:stop(Pid)
+    end,
+    _ = erlang:system_monitor(undefined),
+    ok.
+
+installs_and_releases_system_monitor(_) ->
+    fun() ->
+        {ok, Pid} = gateway_stall_monitor:start_link(),
+        unlink(Pid),
+        {Pid, Opts} = erlang:system_monitor(),
+        ?assertEqual(200, proplists:get_value(long_schedule, Opts)),
+        ?assertEqual(200, proplists:get_value(long_gc, Opts)),
+        ?assertMatch(#{status := installed}, gateway_stall_monitor:snapshot()),
+        ok = gen_server:stop(Pid),
+        ?assertEqual(undefined, erlang:system_monitor())
+    end.
+
+counts_forwarded_events_and_names_the_process(_) ->
+    fun() ->
+        {ok, Pid} = gateway_stall_monitor:start_link(),
+        unlink(Pid),
+        Busy = spawn(fun() ->
+            receive
+                stop -> ok
+            end
+        end),
+        Info = [{timeout, 1500}, {in, {guild, handle_info, 2}}, {out, {guild, handle_info, 2}}],
+        Pid ! {monitor, Busy, long_schedule, Info},
+        Pid ! {monitor, Busy, long_gc, [{timeout, 300}, {heap_size, 10}, {old_heap_size, 0}]},
+        #{stats := Stats, recent := Recent} = gateway_stall_monitor:snapshot(),
+        ?assertEqual(
+            #{count => 1, severe => 1, max_ms => 1500}, maps:get(long_schedule, Stats)
+        ),
+        ?assertEqual(#{count => 1, severe => 0, max_ms => 300}, maps:get(long_gc, Stats)),
+        [GcEvent, SchedEvent] = Recent,
+        ?assertMatch(#{kind := long_gc, ms := 300, info := #{heap_size := 10}}, GcEvent),
+        ?assertMatch(
+            #{kind := long_schedule, who := Busy, info := #{in := {guild, handle_info, 2}}},
+            SchedEvent
+        ),
+        ?assertMatch(#{who_info := #{guild_id := undefined}}, SchedEvent),
+        Busy ! stop
+    end.
+
+leaves_a_foreign_system_monitor_alone(_) ->
+    fun() ->
+        Foreign = spawn(fun() ->
+            receive
+                stop -> ok
+            end
+        end),
+        _ = erlang:system_monitor(Foreign, [{long_gc, 500}]),
+        {ok, Pid} = gateway_stall_monitor:start_link(),
+        unlink(Pid),
+        ?assertMatch(#{status := not_owner}, gateway_stall_monitor:snapshot()),
+        ok = gen_server:stop(Pid),
+        ?assertMatch({Foreign, _}, erlang:system_monitor()),
+        Foreign ! stop
+    end.
+
+records_late_timer_ticks(_) ->
+    fun() ->
+        {ok, Pid} = gateway_stall_monitor:start_link(),
+        unlink(Pid),
+        _ = sys:replace_state(Pid, fun(S) -> S#{tick_due := maps:get(tick_due, S) - 1300} end),
+        Pid ! tick,
+        #{stats := Stats} = gateway_stall_monitor:snapshot(),
+        #{count := Count, max_ms := Max} = maps:get(timer_late, Stats),
+        ?assert(Count >= 1),
+        ?assert(Max >= 1200)
+    end.

fluxer_gateway/test/guild_channel_move_equivalence_tests.erl

@@ -479,6 +479,76 @@ stale_detection_tracks_every_permission_input_test_() ->
         end)
     end}.
 
+twin_channel_engines_match_a_fresh_build_test_() ->
+    {timeout, 120, fun() ->
+        with_harness(fun() ->
+            {State0, Env} = materialize(twin_spec()),
+            Lists = twin_lists(),
+            State = lists:foldl(fun guild_member_list_channel_engine:ensure/2, State0, Lists),
+            Refs = [guild_member_list_channel_engine:ref(L, State) || L <- Lists],
+            ?assertEqual(length(Lists), length(lists:usort(Refs))),
+            Final = lists:foldl(fun assert_engine_matches_fresh_build/2, State, Lists),
+            teardown(Final, Env)
+        end)
+    end}.
+
+stale_twin_engine_is_not_cloned_test_() ->
+    {timeout, 120, fun() ->
+        with_harness(fun() ->
+            {State0, Env} = materialize(twin_spec()),
+            [A, B | _] = twin_lists(),
+            State1 = guild_member_list_channel_engine:ensure(A, State0),
+            Hidden = with_channels(
+                fun(C) ->
+                    case integer_to_binary(channel_int_id(C)) of
+                        A -> C#{<<"permission_overwrites">> => [hidden_overwrite()]};
+                        _ -> C
+                    end
+                end,
+                State1
+            ),
+            [
+                ok = guild_member_list_channel_engine:update_user(user_id(I), A, Hidden)
+             || I <- lists:seq(1, 60)
+            ],
+            State2 = guild_member_list_channel_engine:ensure(B, Hidden),
+            ?assert(guild_member_list_engine_inputs:is_stale(A, State2)),
+            ?assertNotEqual(engine_content(A, State2), engine_content(B, State2)),
+            teardown(assert_engine_matches_fresh_build(B, State2), Env)
+        end)
+    end}.
+
+twin_spec() ->
+    Restricted = [overwrite(role_id(1), 0, 0, view())],
+    single_session_spec(
+        renumber([
+            channel(?CHAN_BASE + 700, 0, null, []),
+            channel(?CHAN_BASE + 701, 0, null, []),
+            channel(?CHAN_BASE + 702, 0, null, Restricted),
+            channel(?CHAN_BASE + 703, 0, null, Restricted)
+        ])
+    ).
+
+twin_lists() ->
+    [integer_to_binary(?CHAN_BASE + I) || I <- [700, 701, 702, 703]].
+
+assert_engine_matches_fresh_build(ListId, State) ->
+    Built = engine_content(ListId, State),
+    Rebuilt = guild_member_list_channel_engine:rebuild(ListId, State),
+    ?assertEqual(engine_content(ListId, Rebuilt), Built),
+    Rebuilt.
+
+engine_content(ListId, State) ->
+    Ref = guild_member_list_channel_engine:ref(ListId, State),
+    Items = guild_member_list_engine:get_all_item_keys(Ref),
+    Members = [UserId || {member, UserId} <- Items],
+    {
+        guild_member_list_engine:get_counts(Ref),
+        guild_member_list_engine:get_groups(Ref),
+        Items,
+        [{U, guild_member_list_engine:is_member_online(Ref, U)} || U <- Members]
+    }.
+
 input_changes() ->
     Flip = flip_channel(),
     [

fluxer_gateway/test/guild_connect_async_tests.erl

@@ -123,6 +123,132 @@ enqueue_session_connect_async_allows_immediate_start_when_wait_queue_disabled_te
     WorkerRefs = maps:get(session_connect_worker_refs, State1, #{}),
     cleanup_worker_refs(WorkerRefs).
 
+finalize_batch_upserts_every_result_and_frees_one_worker_test() ->
+    UserId = 10,
+    Sessions = [<<"s1">>, <<"s2">>],
+    State0 = (finalize_state(<<"s1">>, UserId, 1, true, #{}, sets:new(), #{UserId => 1}))#{
+        sessions => maps:from_list([{S, pending_entry(S, UserId)} || S <- Sessions]),
+        session_connect_pending => maps:from_list([{S, 1} || S <- Sessions]),
+        session_connect_inflight => 2
+    },
+    State1 = guild_connect_async:finalize_session_connect_batch(
+        [{S, 1, {ok, #{}}, finalize_computed(S, UserId)} || S <- Sessions], State0
+    ),
+    ?assertEqual(1, maps:get(session_connect_inflight, State1)),
+    ?assertEqual(#{}, maps:get(session_connect_pending, State1)),
+    [
+        ?assertEqual(false, maps:get(pending_connect, maps:get(S, maps:get(sessions, State1))))
+     || S <- Sessions
+    ],
+    ?assertEqual(#{UserId => 2}, maps:get(user_session_counts, State1)),
+    ?assertEqual(3, maps:get(UserId, maps:get(presence_subscriptions, State1))),
+    ?assertEqual(2, length(flush_connect_results())).
+
+finalize_batch_skips_stale_attempts_test() ->
+    UserId = 11,
+    State0 = finalize_state(<<"s1">>, UserId, 2, true, #{}, sets:new(), #{UserId => 1}),
+    State1 = guild_connect_async:finalize_session_connect_batch(
+        [{<<"s1">>, 1, {ok, #{}}, finalize_computed(<<"s1">>, UserId)}], State0
+    ),
+    ?assertEqual(#{<<"s1">> => 2}, maps:get(session_connect_pending, State1)),
+    ?assertEqual(
+        true, maps:get(pending_connect, maps:get(<<"s1">>, maps:get(sessions, State1)))
+    ),
+    ?assertEqual([], flush_connect_results()).
+
+full_guild_reconnect_fits_the_connect_queue_test() ->
+    GuildId = 42,
+    State0 = (saturated_connect_state(GuildId, <<"s-0">>, 0, 1))#{
+        session_connect_queue => queue:new(),
+        session_connect_pending => #{},
+        sessions => #{}
+    },
+    State1 = lists:foldl(
+        fun(N, Acc) ->
+            SessionId = integer_to_binary(N),
+            guild_connect_async:enqueue_session_connect_async(
+                GuildId,
+                0,
+                connect_request(SessionId, N),
+                #{},
+                maps:remove(session_connect_max_queue, Acc)
+            )
+        end,
+        State0,
+        lists:seq(1, 3000)
+    ),
+    ?assertEqual(3000, queue:len(maps:get(session_connect_queue, State1))),
+    ?assertEqual(3000, map_size(maps:get(session_connect_pending, State1))),
+    receive
+        {guild_connect_result, GuildId, _, {error, overloaded}} ->
+            ?assert(false, connect_dropped)
+    after 0 ->
+        ok
+    end.
+
+queued_connects_spread_over_batched_workers_test() ->
+    ok = flush_batches(),
+    GuildId = 42,
+    Requests = [connect_request(integer_to_binary(N), N) || N <- lists:seq(1, 20)],
+    State0 = #{
+        id => GuildId,
+        sessions => #{},
+        session_connect_queue => queue:from_list([
+            #{guild_id => GuildId, attempt => 0, request => R, reply_via_pid => undefined}
+         || R <- Requests
+        ]),
+        session_connect_pending => #{},
+        session_connect_inflight => 0,
+        data => #{},
+        member_count => 0,
+        voice_states => #{},
+        member_list_engine => undefined,
+        virtual_channel_access => #{}
+    },
+    State1 = guild_connect_async:maybe_start_session_connect_workers(State0),
+    WorkerRefs = maps:get(session_connect_worker_refs, State1),
+    ?assertEqual(8, maps:get(session_connect_inflight, State1)),
+    ?assertEqual(8, map_size(WorkerRefs)),
+    ?assertEqual(0, queue:len(maps:get(session_connect_queue, State1))),
+    Batches = [receive_batch() || _ <- lists:seq(1, 8)],
+    ?assertEqual(
+        lists:sort([maps:get(session_id, R) || R <- Requests]),
+        lists:sort([SessionId || Batch <- Batches, {SessionId, 0, _, _} <- Batch])
+    ),
+    cleanup_worker_refs(WorkerRefs).
+
+pending_entry(SessionId, UserId) ->
+    #{
+        session_id => SessionId,
+        user_id => UserId,
+        pid => self(),
+        mref => make_ref(),
+        pending_connect => true,
+        active_guilds => sets:new()
+    }.
+
+receive_batch() ->
+    receive
+        {'$gen_cast', {session_connect_worker_batch_done, Results}} -> Results
+    after 5000 ->
+        ?assert(false, connect_batch_not_received)
+    end.
+
+flush_batches() ->
+    receive
+        {'$gen_cast', {session_connect_worker_batch_done, _Results}} -> flush_batches()
+    after 100 ->
+        ok
+    end.
+
+flush_connect_results() ->
+    receive
+        {guild_connect_result, _GuildId, _Attempt, _Reply} = Msg ->
+            [Msg | flush_connect_results()]
+    after 0 ->
+        []
+    end.
+
 finalize_state(SessionId, UserId, Attempt, PendingConnect, Counts, Connected, PresenceSubs) ->
     Existing = #{
         session_id => SessionId,

fluxer_gateway/test/guild_data_tests.erl

@@ -81,6 +81,35 @@ read_model_observes_role_and_collection_changes_test() ->
         cleanup_read_model(State)
     end.
 
+read_model_observes_last_message_and_pin_advances_test() ->
+    State = read_model_state(),
+    try
+        ok = guild_read_model:put_state(State),
+        Data = maps:get(data, State),
+        Advanced = guild_state_channels:handle_message_create(
+            #{<<"channel_id">> => <<"500">>, <<"id">> => <<"900">>}, Data
+        ),
+        Pinned = guild_state_channels:handle_channel_pins_update(
+            #{
+                <<"channel_id">> => <<"500">>,
+                <<"last_pin_timestamp">> => <<"2026-10-02T00:00:00Z">>
+            },
+            Advanced
+        ),
+        Updated = State#{data => Pinned},
+        ok = guild_read_model:update(State, Updated),
+        {reply, Expected, _} = guild_data:get_guild_data(#{user_id => 200}, Updated),
+        ?assertEqual(
+            {ok, Expected}, guild_read_model:query(100, {get_guild_data, #{user_id => 200}})
+        ),
+        #{guild_data := #{<<"channels">> := Channels}} = Expected,
+        [Channel] = [C || C <- Channels, maps:get(<<"id">>, C) =:= 500],
+        ?assertEqual(900, maps:get(<<"last_message_id">>, Channel)),
+        ?assertEqual(<<"2026-10-02T00:00:00Z">>, maps:get(<<"last_pin_timestamp">>, Channel))
+    after
+        cleanup_read_model(State)
+    end.
+
 read_model_survives_blocked_owner_and_rejects_dead_tables_test() ->
     State = read_model_state(),
     Self = self(),

fluxer_gateway/test/guild_dm_partners_tests.erl

@@ -83,6 +83,17 @@ guild_wide_visibility_change_reevaluates_test() ->
         ?assertEqual({dm_partner_mutual, ?GUILD_ID, [30]}, receive_mutual())
     end).
 
+partner_role_change_after_registration_reevaluates_test() ->
+    with_flag(true, fun() ->
+        Before = register_partners(state(), [30]),
+        ?assertEqual(none, receive_mutual()),
+        After = with_member(member(30, [?VIEWER_ROLE]), Before),
+        _ = guild_dm_partners:maybe_reevaluate(
+            guild_member_update, #{<<"user">> => #{<<"id">> => <<"30">>}}, Before, After
+        ),
+        ?assertEqual({dm_partner_mutual, ?GUILD_ID, [30]}, receive_mutual())
+    end).
+
 disconnected_sessions_are_dropped_on_reevaluation_test() ->
     with_flag(true, fun() ->
         Before = register_partners(state(), [20]),
@@ -136,6 +147,11 @@ members() ->
 without_member(UserId, #{data := Data} = State) ->
     State#{data => Data#{<<"members">> => maps:remove(UserId, maps:get(<<"members">>, Data))}}.
 
+with_member(Member, #{data := Data} = State) ->
+    #{<<"user">> := #{<<"id">> := RawId}} = Member,
+    Members = maps:get(<<"members">>, Data),
+    State#{data => Data#{<<"members">> => Members#{binary_to_integer(RawId) => Member}}}.
+
 with_channels(Channels, #{data := Data} = State) ->
     State#{
         data => Data#{

fluxer_gateway/test/guild_handoff_freeze_tests.erl

@@ -0,0 +1,468 @@
+%% SPDX-License-Identifier: AGPL-3.0-or-later
+
+-module(guild_handoff_freeze_tests).
+-behaviour(gen_server).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
+
+-define(GUILD_ID, 4242).
+-define(SHARD_TABLE, guild_manager_shard_table).
+-define(SINK, guild_handoff_freeze_tests_sink).
+
+transfer_test_() ->
+    {foreach, fun setup/0, fun cleanup/1, [
+        instantiate(fun frozen_backlog_reaches_target_and_source_applies_nothing/1),
+        instantiate(fun final_window_call_fails_unapplied_and_cast_is_kept/1),
+        instantiate(fun route_failure_restores_routes_resumes_source_and_stops_target/1),
+        instantiate(fun controller_death_before_start_resumes_source/1),
+        instantiate(fun controller_death_after_start_kills_target_and_resumes_source/1),
+        instantiate(fun target_death_before_commit_keeps_source/1),
+        instantiate(fun sessions_stay_behind_when_not_transferred/1),
+        instantiate(fun cast_during_source_terminate_reaches_target/1),
+        instantiate(fun monitor_flush_in_source_terminate_loses_no_cast/1),
+        instantiate(fun connect_worker_results_stay_with_the_source/1),
+        instantiate(fun abort_after_forwarding_reports_the_forwarded_count/1),
+        instantiate(fun late_source_stop_keeps_the_committed_target/1),
+        instantiate(fun controller_death_after_commit_keeps_the_target/1),
+        instantiate(fun lost_guard_aborts_before_other_nodes_route_to_the_target/1)
+    ]}.
+
+instantiate(Test) ->
+    fun(Ctx) -> {timeout, 30, fun() -> Test(Ctx) end} end.
+
+frozen_backlog_reaches_target_and_source_applies_nothing(#{src := Src, src_shard := SrcShard}) ->
+    Test = self(),
+    BeforeStart = fun() ->
+        gen_server:cast(Src, {append, b}),
+        spawn(fun() -> Test ! {call_reply, gen_server:call(Src, {append_call, c}, 30000)} end),
+        wait_mailbox(Src, 2)
+    end,
+    Result = guild_handoff_freeze:transfer(
+        ?GUILD_ID, Src, SrcShard, node(), #{before_start => BeforeStart}
+    ),
+    ?assertMatch({ok, #{new_pid := _}}, Result),
+    {ok, #{new_pid := NewPid}} = Result,
+    ?assertEqual({call_reply, {ok, NewPid}}, receive_tagged(call_reply)),
+    ?assertEqual([{Src, {shutdown, handoff}, [a]}], ets:lookup(?SINK, Src)),
+    ?assertEqual([a, b, c], gen_server:call(NewPid, get_log)),
+    ?assertEqual(false, is_process_alive(Src)).
+
+final_window_call_fails_unapplied_and_cast_is_kept(#{src := Src, src_shard := SrcShard}) ->
+    Test = self(),
+    BeforeStart = fun() ->
+        gen_server:cast(Src, {append, b}),
+        spawn(fun() -> Test ! {call_reply, call_or_exit(Src, {append_call, c})} end),
+        wait_mailbox(Src, 2)
+    end,
+    Opts = #{before_start => BeforeStart, forward_rounds => 0},
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts),
+    ?assertMatch({ok, #{final := #{cast := 1, dropped_call := 1}}}, Result),
+    {ok, #{new_pid := NewPid}} = Result,
+    ?assertMatch({call_reply, {exit, {{shutdown, handoff}, _}}}, receive_tagged(call_reply)),
+    ?assertEqual([a, b], gen_server:call(NewPid, get_log)).
+
+route_failure_restores_routes_resumes_source_and_stops_target(#{
+    src := Src, src_shard := SrcShard, dst_shard := DstShard
+}) ->
+    Test = self(),
+    Opts = #{
+        before_start => fun() -> gen_server:cast(Src, {append, b}) end,
+        after_start => fun(_NewPid) -> {error, route_rejected} end,
+        on_abort => fun() ->
+            Test ! routes_restored,
+            ok
+        end
+    },
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts),
+    ?assertMatch({error, #{phase := after_start, reason := route_rejected}}, Result),
+    ?assertEqual(routes_restored, receive_tagged(routes_restored)),
+    ?assertEqual({error, not_found}, gen_server:call(DstShard, {lookup, ?GUILD_ID})),
+    ?assertEqual([a, b], gen_server:call(Src, get_log)),
+    ?assertEqual(false, guild_handoff_freeze:is_frozen(Src)).
+
+controller_death_before_start_resumes_source(#{src := Src, src_shard := SrcShard}) ->
+    Opts = #{
+        before_start => fun() ->
+            gen_server:cast(Src, {append, b}),
+            exit(self(), kill)
+        end
+    },
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts),
+    ?assertMatch({error, #{phase := crashed, reason := killed}}, Result),
+    ?assertEqual([a, b], gen_server:call(Src, get_log, 5000)).
+
+controller_death_after_start_kills_target_and_resumes_source(#{
+    src := Src, src_shard := SrcShard, dst_shard := DstShard
+}) ->
+    Test = self(),
+    Opts = #{
+        after_start => fun(NewPid) ->
+            Test ! {target, NewPid},
+            exit(self(), kill)
+        end,
+        on_abort => fun() ->
+            Test ! routes_restored,
+            ok
+        end
+    },
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts),
+    ?assertMatch({error, #{phase := crashed, abort := #{routes := ok, target := ok}}}, Result),
+    ?assertEqual(routes_restored, receive_tagged(routes_restored)),
+    {target, NewPid} = receive_tagged(target),
+    ?assertEqual(ok, wait_dead(NewPid)),
+    ?assertEqual({error, not_found}, gen_server:call(DstShard, {lookup, ?GUILD_ID})),
+    ?assertEqual([a], gen_server:call(Src, get_log, 5000)).
+
+target_death_before_commit_keeps_source(#{src := Src, src_shard := SrcShard}) ->
+    Opts = #{
+        after_start => fun(NewPid) ->
+            exit(NewPid, kill),
+            wait_dead(NewPid)
+        end
+    },
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts),
+    ?assertMatch({error, #{phase := target_died, reason := killed}}, Result),
+    ?assertEqual([a], gen_server:call(Src, get_log, 5000)),
+    ?assertEqual(false, guild_handoff_freeze:is_frozen(Src)).
+
+sessions_stay_behind_when_not_transferred(#{src := Src, src_shard := SrcShard}) ->
+    ok = gen_server:call(Src, {put_session, <<"s1">>, #{pid => self(), user_id => 7}}),
+    Opts = #{transfer_sessions => false},
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts),
+    ?assertMatch({ok, #{new_pid := _}}, Result),
+    {ok, #{new_pid := NewPid}} = Result,
+    ?assertEqual(#{}, gen_server:call(NewPid, get_sessions)),
+    ?assertEqual([a], gen_server:call(NewPid, get_log)).
+
+cast_during_source_terminate_reaches_target(#{src := Src, src_shard := SrcShard}) ->
+    ok = gen_server:call(Src, cast_late_on_terminate),
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), #{}),
+    ?assertMatch({ok, #{final := #{cast := 1}}}, Result),
+    {ok, #{new_pid := NewPid}} = Result,
+    ?assertEqual([a, late], gen_server:call(NewPid, get_log)).
+
+monitor_flush_in_source_terminate_loses_no_cast(#{src := Src, src_shard := SrcShard}) ->
+    ok = gen_server:call(Src, flush_down_on_terminate),
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), #{}),
+    ?assertMatch({ok, #{final := #{cast := 1}}}, Result),
+    {ok, #{new_pid := NewPid}} = Result,
+    ?assertEqual([a, late], gen_server:call(NewPid, get_log)).
+
+connect_worker_results_stay_with_the_source(#{src := Src, src_shard := SrcShard}) ->
+    BeforeStart = fun() ->
+        gen_server:cast(Src, {session_connect_worker_done, <<"s1">>, 1, {ok, #{}}, #{}}),
+        gen_server:cast(
+            Src, {session_connect_worker_batch_done, [{<<"s1">>, 1, {ok, #{}}, #{}}]}
+        ),
+        gen_server:cast(Src, {append, b}),
+        wait_mailbox(Src, 3)
+    end,
+    Result = guild_handoff_freeze:transfer(
+        ?GUILD_ID, Src, SrcShard, node(), #{before_start => BeforeStart}
+    ),
+    ?assertMatch({ok, #{new_pid := _}}, Result),
+    {ok, #{new_pid := NewPid}} = Result,
+    ?assertEqual([a, b], gen_server:call(NewPid, get_log)).
+
+abort_after_forwarding_reports_the_forwarded_count(#{src := Src, src_shard := SrcShard}) ->
+    BeforeStart = fun() ->
+        gen_server:cast(Src, crash),
+        wait_mailbox(Src, 1)
+    end,
+    Result = guild_handoff_freeze:transfer(
+        ?GUILD_ID, Src, SrcShard, node(), #{before_start => BeforeStart}
+    ),
+    ?assertMatch(
+        {error, #{phase := target_died, abort := #{forwarded := 1, thaw := ok}}}, Result
+    ).
+
+late_source_stop_keeps_the_committed_target(#{
+    src := Src, src_shard := SrcShard, dst_shard := DstShard
+}) ->
+    Opts = #{
+        stop_timeout => 100,
+        after_start => fun(_NewPid) -> hold(SrcShard, 600) end
+    },
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts),
+    ?assertMatch({ok, #{new_pid := _}}, Result),
+    {ok, #{new_pid := NewPid}} = Result,
+    ?assertEqual(ok, wait_dead(Src)),
+    ?assert(is_process_alive(NewPid)),
+    ?assertEqual({ok, NewPid}, gen_server:call(DstShard, {lookup, ?GUILD_ID})),
+    ?assertEqual([a], gen_server:call(NewPid, get_log)).
+
+controller_death_after_commit_keeps_the_target(#{
+    src := Src, src_shard := SrcShard, dst_shard := DstShard
+}) ->
+    Test = self(),
+    Opts = #{
+        after_start => fun(NewPid) ->
+            Test ! {started, self(), NewPid},
+            hold(SrcShard, 1000)
+        end
+    },
+    spawn(fun() ->
+        Test ! {result, guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts)}
+    end),
+    {started, Controller, NewPid} = receive_tagged(started),
+    ok = wait_mailbox(SrcShard, 1),
+    exit(Controller, kill),
+    {result, Result} = receive_tagged(result),
+    ?assertMatch({error, #{phase := crashed, committed := true}}, Result),
+    ?assertEqual(ok, wait_dead(Src)),
+    ?assert(is_process_alive(NewPid)),
+    ?assertEqual({ok, NewPid}, gen_server:call(DstShard, {lookup, ?GUILD_ID})).
+
+lost_guard_aborts_before_other_nodes_route_to_the_target(#{
+    src := Src, src_shard := SrcShard, dst_shard := DstShard
+}) ->
+    Test = self(),
+    Opts = #{
+        before_start => fun() ->
+            {monitored_by, Watchers} = process_info(self(), monitored_by),
+            [exit(W, kill) || W <- Watchers, W =/= Test],
+            ok
+        end,
+        after_start => fun(_NewPid) ->
+            Test ! routed,
+            ok
+        end
+    },
+    Result = guild_handoff_freeze:transfer(?GUILD_ID, Src, SrcShard, node(), Opts),
+    ?assertMatch(
+        {error, #{
+            phase := after_start, reason := {guard_lost, killed}, abort := #{exposed := false}
+        }},
+        Result
+    ),
+    ?assertEqual({timeout, routed}, receive_tagged(routed, 200)),
+    ?assertEqual({error, not_found}, gen_server:call(DstShard, {lookup, ?GUILD_ID})),
+    ?assertEqual(false, guild_handoff_freeze:is_frozen(Src)),
+    ?assertEqual([a], gen_server:call(Src, get_log, 5000)).
+
+hold(Pid, Ms) ->
+    Test = self(),
+    spawn(fun() ->
+        true = erlang:suspend_process(Pid),
+        Test ! {held, Pid},
+        timer:sleep(Ms),
+        true = erlang:resume_process(Pid)
+    end),
+    receive
+        {held, Pid} -> ok
+    end.
+
+export_drops_only_keys_the_importer_rebuilds_test() ->
+    Raw = #{
+        <<"guild">> => #{<<"id">> => <<"4242">>},
+        <<"roles">> => [#{<<"id">> => <<"4242">>, <<"permissions">> => <<"1024">>}],
+        <<"channels">> => [#{<<"id">> => <<"77">>, <<"type">> => 0}],
+        <<"members">> => [
+            #{<<"user">> => #{<<"id">> => <<"5">>}, <<"roles">> => []},
+            #{<<"user">> => #{<<"id">> => <<"6">>}, <<"roles">> => [<<"4242">>]}
+        ]
+    },
+    Full = guild_data_index:normalize_map(Raw),
+    State = #{id => ?GUILD_ID, data => Full#{members_ets => make_ref()}, sessions => #{}},
+    Slim = maps:get(data, guild_handoff:export_handoff_state(State)),
+    ?assertEqual([], [K || K <- guild_handoff:derived_data_keys(), maps:is_key(K, Slim)]),
+    Rebuild = fun(D) ->
+        maps:remove(member_list_revision, guild_data_index:normalize_map(D))
+    end,
+    ?assertEqual(Rebuild(Full), Rebuild(Slim)).
+
+export_keeps_channel_fields_that_only_the_index_holds_test() ->
+    Raw = #{
+        <<"guild">> => #{<<"id">> => <<"4242">>},
+        <<"channels">> => [
+            #{<<"id">> => <<"77">>, <<"type">> => 0, <<"last_message_id">> => <<"10">>}
+        ]
+    },
+    Full = guild_state_channels:handle_message_create(
+        #{<<"channel_id">> => <<"77">>, <<"id">> => <<"900">>},
+        guild_data_index:normalize_map(Raw)
+    ),
+    State = #{id => ?GUILD_ID, data => Full, sessions => #{}},
+    Slim = maps:get(data, guild_handoff:export_handoff_state(State)),
+    ?assertMatch({rebuilt, _}, bounded_rebuild(Slim)),
+    {rebuilt, Rebuilt} = bounded_rebuild(Slim),
+    [Channel] = guild_data_index:channel_list(Rebuilt),
+    ?assertEqual(900, maps:get(<<"last_message_id">>, Channel)).
+
+bounded_rebuild(Data) ->
+    {Pid, Ref} = spawn_monitor(fun() ->
+        process_flag(max_heap_size, #{size => 4000000, kill => true, error_logger => false}),
+        exit({rebuilt, guild_data_index:normalize_map(Data)})
+    end),
+    receive
+        {'DOWN', Ref, process, Pid, Reason} -> Reason
+    after 10000 ->
+        exit(Pid, kill),
+        timeout
+    end.
+
+setup() ->
+    ets:new(?SHARD_TABLE, [named_table, public, set]),
+    ets:new(?SINK, [named_table, public, set]),
+    {ok, SrcShard} = gen_server:start(?MODULE, shard, []),
+    {ok, DstShard} = gen_server:start(?MODULE, shard, []),
+    Index = guild_manager_shards:select_shard(?GUILD_ID, 1),
+    ets:insert(?SHARD_TABLE, [{shard_count, 1}, {{shard_pid, Index}, DstShard}]),
+    {ok, Src} = gen_server:call(
+        SrcShard, {start_transferred, ?GUILD_ID, guild_export([a])}
+    ),
+    #{src => Src, src_shard => SrcShard, dst_shard => DstShard}.
+
+cleanup(#{src_shard := SrcShard, dst_shard := DstShard}) ->
+    Guilds = lists:append([gen_server:call(S, all_guilds) || S <- [SrcShard, DstShard]]),
+    [exit(G, kill) || G <- Guilds],
+    [gen_server:stop(S) || S <- [SrcShard, DstShard]],
+    ets:delete(?SHARD_TABLE),
+    ets:delete(?SINK),
+    flush().
+
+guild_export(Log) ->
+    #{
+        id => ?GUILD_ID,
+        data => #{<<"log">> => Log},
+        sessions => #{},
+        voice_states => #{}
+    }.
+
+init(shard) ->
+    {ok, #{role => shard, guilds => #{}}};
+init({guild, Export}) ->
+    process_flag(trap_exit, true),
+    {ok, Export#{role => guild}}.
+
+handle_call({start_transferred, GuildId, Export}, _From, #{role := shard} = State) ->
+    {ok, Pid} = gen_server:start(?MODULE, {guild, Export}, []),
+    {reply, {ok, Pid}, put_guild(GuildId, Pid, State)};
+handle_call({stop_guild, GuildId}, From, #{role := shard} = State) ->
+    handle_call({stop_guild, GuildId, normal}, From, State);
+handle_call({stop_guild, GuildId, Reason}, _From, #{role := shard, guilds := Guilds} = State) ->
+    case maps:get(GuildId, Guilds, undefined) of
+        Pid when is_pid(Pid) -> stop_quietly(Pid, Reason);
+        undefined -> ok
+    end,
+    {reply, ok, State#{guilds => maps:remove(GuildId, Guilds)}};
+handle_call({lookup, GuildId}, _From, #{role := shard, guilds := Guilds} = State) ->
+    case maps:get(GuildId, Guilds, undefined) of
+        Pid when is_pid(Pid) ->
+            case is_process_alive(Pid) of
+                true -> {reply, {ok, Pid}, State};
+                false -> {reply, {error, not_found}, State}
+            end;
+        undefined ->
+            {reply, {error, not_found}, State}
+    end;
+handle_call(all_guilds, _From, #{role := shard, guilds := Guilds} = State) ->
+    {reply, maps:values(Guilds), State};
+handle_call({get_guild_id}, _From, #{role := guild, id := Id} = State) ->
+    {reply, Id, State};
+handle_call({put_session, Id, Session}, _From, #{role := guild, sessions := Sessions} = State) ->
+    {reply, ok, State#{sessions => Sessions#{Id => Session}}};
+handle_call(flush_down_on_terminate, _From, #{role := guild} = State) ->
+    Helper = spawn(fun() ->
+        receive
+            stop -> ok
+        end
+    end),
+    {reply, ok, State#{flush_down => {Helper, erlang:monitor(process, Helper)}}};
+handle_call(cast_late_on_terminate, _From, #{role := guild} = State) ->
+    {reply, ok, State#{cast_late => true}};
+handle_call(get_sessions, _From, #{role := guild, sessions := Sessions} = State) ->
+    {reply, Sessions, State};
+handle_call(get_log, _From, #{role := guild} = State) ->
+    {reply, log(State), State};
+handle_call({append_call, Item}, _From, #{role := guild} = State) ->
+    {reply, {ok, self()}, append(Item, State)}.
+
+handle_cast({append, Item}, #{role := guild} = State) ->
+    {noreply, append(Item, State)};
+handle_cast(crash, #{role := guild} = State) ->
+    {stop, crashed, State}.
+
+handle_info({'EXIT', _Pid, Reason}, #{role := guild} = State) ->
+    {stop, Reason, State};
+handle_info(_Msg, State) ->
+    {noreply, State}.
+
+terminate(Reason, #{role := guild} = State) ->
+    ets:insert(?SINK, {self(), Reason, log(State)}),
+    case maps:get(cast_late, State, false) of
+        true ->
+            gen_server:cast(self(), {append, late}),
+            timer:sleep(200);
+        false ->
+            ok
+    end,
+    case maps:get(flush_down, State, undefined) of
+        {Helper, Ref} ->
+            exit(Helper, kill),
+            timer:sleep(100),
+            erlang:demonitor(Ref, [flush]),
+            gen_server:cast(self(), {append, late}),
+            timer:sleep(200);
+        undefined ->
+            ok
+    end;
+terminate(_Reason, _State) ->
+    ok.
+
+put_guild(GuildId, Pid, #{guilds := Guilds} = State) ->
+    State#{guilds => Guilds#{GuildId => Pid}}.
+
+log(#{data := Data}) ->
+    maps:get(<<"log">>, Data).
+
+append(Item, #{data := Data} = State) ->
+    State#{data => Data#{<<"log">> => log(State) ++ [Item]}}.
+
+call_or_exit(Pid, Request) ->
+    try
+        gen_server:call(Pid, Request, 30000)
+    catch
+        exit:Reason -> {exit, Reason}
+    end.
+
+stop_quietly(Pid, Reason) ->
+    try
+        gen_server:stop(Pid, Reason, 5000)
+    catch
+        exit:_ -> ok
+    end.
+
+wait_mailbox(Pid, N) ->
+    case process_info(Pid, message_queue_len) of
+        {message_queue_len, Len} when Len >= N ->
+            ok;
+        _ ->
+            timer:sleep(5),
+            wait_mailbox(Pid, N)
+    end.
+
+wait_dead(Pid) ->
+    Ref = erlang:monitor(process, Pid),
+    receive
+        {'DOWN', Ref, process, Pid, _} -> ok
+    after 5000 -> still_alive
+    end.
+
+receive_tagged(Tag) ->
+    receive_tagged(Tag, 10000).
+
+receive_tagged(Tag, Timeout) ->
+    receive
+        Msg when element(1, Msg) =:= Tag -> Msg;
+        Tag -> Tag
+    after Timeout -> {timeout, Tag}
+    end.
+
+flush() ->
+    receive
+        _ -> flush()
+    after 0 -> ok
+    end.

fluxer_gateway/test/guild_handoff_tests.erl

@@ -36,7 +36,13 @@ restore_transferred_session_state_rebuilds_connected_counts_test() ->
         <<"s1">> => #{pid => self(), user_id => UserId},
         <<"s2">> => #{pid => self(), user_id => UserId, pending_connect => false},
         <<"pending">> => #{pid => self(), user_id => 11, pending_connect => true},
-        <<"bad">> => #{user_id => 12}
+        <<"bad">> => #{user_id => 12},
+        <<"reconnecting">> => #{
+            pid => self(),
+            user_id => 13,
+            pending_connect => true,
+            owns_connected_tracking => true
+        }
     },
     State0 = #{
         sessions => Sessions,
@@ -45,10 +51,11 @@ restore_transferred_session_state_rebuilds_connected_counts_test() ->
         presence_subscriptions => #{UserId => 1}
     },
     State1 = guild_handoff:restore_transferred_session_state(State0),
-    ?assertEqual(#{UserId => 2}, maps:get(user_session_counts, State1)),
+    ?assertEqual(#{UserId => 2, 13 => 1}, maps:get(user_session_counts, State1)),
     Connected = maps:get(connected_user_ids, State1),
     ?assertEqual(true, sets:is_element(UserId, Connected)),
     ?assertEqual(false, sets:is_element(11, Connected)),
+    ?assertEqual(true, sets:is_element(13, Connected)),
     ?assertEqual(3, maps:get(UserId, maps:get(presence_subscriptions, State1))).
 
 export_handoff_state_contains_all_required_keys_test() ->

fluxer_gateway/test/guild_manager_tests.erl

@@ -69,6 +69,35 @@ handoff_guild_ids_counts_attempts_and_successes_test() ->
     ?assertEqual(#{attempted => 2, handed_off => 1}, Result),
     ?assertEqual(3, maps:get(shard_count, FinalState)).
 
+handoff_to_topology_keeps_guild_the_router_owns_here_test() ->
+    GuildId = 77,
+    ShardPid = spawn(fun() -> local_ids_stub_loop([GuildId]) end),
+    State = #{shards => #{0 => #{pid => ShardPid, ref => make_ref()}}, shard_count => 1},
+    persistent_term:put({gateway_cluster_membership, members}, [node()]),
+    persistent_term:put({gateway_cluster_membership, members_by_role}, #{guilds => [node()]}),
+    try
+        {Result, _State} = guild_manager_handoff:perform_handoff_to_topology(
+            ['[email protected]'], State
+        ),
+        ?assertEqual(#{attempted => 0, handed_off => 0}, Result)
+    after
+        ShardPid ! stop,
+        persistent_term:erase({gateway_cluster_membership, members}),
+        persistent_term:erase({gateway_cluster_membership, members_by_role})
+    end.
+
+local_ids_stub_loop(GuildIds) ->
+    receive
+        stop ->
+            ok;
+        {'$gen_call', From, get_local_guild_ids} ->
+            gen_server:reply(From, {ok, GuildIds}),
+            local_ids_stub_loop(GuildIds);
+        {'$gen_call', From, _Request} ->
+            gen_server:reply(From, {error, not_found}),
+            local_ids_stub_loop(GuildIds)
+    end.
+
 find_shard_by_ref_found_test() ->
     Ref = make_ref(),
     Shards = #{0 => #{pid => self(), ref => Ref}},

fluxer_gateway/test/guild_member_list_chaos_tests.erl

@@ -153,8 +153,7 @@ subscribe_and_drain(GuildPid, Requests) ->
         end,
         Requests
     ),
-    GuildPid ! flush_lazy_subscribe_buffer,
-    SubscribedState = get_guild_state(GuildPid),
+    SubscribedState = drain_lazy_subscribe_buffer(GuildPid),
     assert_requests_subscribed(Requests, SubscribedState),
     await_initial_syncs(Requests, SubscribedState),
     _ = collect_captures(100),
@@ -259,6 +258,14 @@ presence_update(GuildPid, UserId, Status) ->
     _ = gen_server:call(GuildPid, {get_counts}, 60000),
     ok.
 
+drain_lazy_subscribe_buffer(GuildPid) ->
+    GuildPid ! flush_lazy_subscribe_buffer,
+    State = get_guild_state(GuildPid),
+    case maps:is_key(lazy_subscribe_buffer, State) of
+        true -> drain_lazy_subscribe_buffer(GuildPid);
+        false -> State
+    end.
+
 get_guild_state(GuildPid) ->
     gen_server:call(GuildPid, {get_sessions}, 60000).
 

fluxer_gateway/test/guild_session_tracking_tests.erl

@@ -0,0 +1,396 @@
+%% SPDX-License-Identifier: AGPL-3.0-or-later
+
+-module(guild_session_tracking_tests).
+-typing([eqwalizer]).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(GUILD_ID, 910100).
+-define(CHANNEL_ID, 910500).
+-define(USER_A, 810001).
+-define(USER_B, 810002).
+-define(USER_C, 810003).
+
+handoff_reconnect_counts_each_session_once_test_() ->
+    {timeout, 60, fun handoff_reconnect_counts_each_session_once/0}.
+
+handoff_reconnect_overflow_keeps_tracking_test_() ->
+    {timeout, 60, fun handoff_reconnect_overflow_keeps_tracking/0}.
+
+reconnect_of_connected_session_counts_once_test_() ->
+    {timeout, 60, fun reconnect_of_connected_session_counts_once/0}.
+
+not_member_reconnect_releases_tracking_test_() ->
+    {timeout, 60, fun not_member_reconnect_releases_tracking/0}.
+
+reconnecting_session_down_releases_tracking_test_() ->
+    {timeout, 60, fun reconnecting_session_down_releases_tracking/0}.
+
+fresh_pending_down_keeps_reconnecting_owner_tracked_test_() ->
+    {timeout, 60, fun fresh_pending_down_keeps_reconnecting_owner_tracked/0}.
+
+unavailable_removal_of_fresh_pending_keeps_owner_tracked_test_() ->
+    {timeout, 60, fun unavailable_removal_of_fresh_pending_keeps_owner_tracked/0}.
+
+fresh_pending_sibling_keeps_mutual_members_test_() ->
+    {timeout, 60, fun fresh_pending_sibling_keeps_mutual_members/0}.
+
+handoff_reconnect_counts_each_session_once() ->
+    with_sessions(
+        [{<<"a1">>, ?USER_A}, {<<"a2">>, ?USER_A}, {<<"b1">>, ?USER_B}],
+        fun(Sessions) ->
+            with_guild(base_state(Sessions), fun(Source) ->
+                await_tracking(#{?USER_A => 2, ?USER_B => 1}, Source),
+                {ok, Exported} = gen_server:call(Source, export_handoff_state, 10000),
+                with_guild(transferred_state(Exported), fun(Target) ->
+                    await_tracking(#{?USER_A => 2, ?USER_B => 1}, Target),
+                    reconnect_all(Target, Sessions, 0),
+                    await_tracking(#{?USER_A => 2, ?USER_B => 1}, Target),
+                    kill_session(<<"b1">>, Sessions),
+                    await_tracking(#{?USER_A => 2}, Target),
+                    kill_session(<<"a1">>, Sessions),
+                    await_tracking(#{?USER_A => 1}, Target),
+                    kill_session(<<"a2">>, Sessions),
+                    await_tracking(#{}, Target)
+                end)
+            end)
+        end
+    ).
+
+handoff_reconnect_overflow_keeps_tracking() ->
+    with_sessions(
+        [{<<"a1">>, ?USER_A}, {<<"b1">>, ?USER_B}, {<<"c1">>, ?USER_C}],
+        fun(Sessions) ->
+            with_guild(base_state(Sessions), fun(Source) ->
+                {ok, Exported} = gen_server:call(Source, export_handoff_state, 10000),
+                with_guild(transferred_state(Exported), fun(Target) ->
+                    Expected = #{?USER_A => 1, ?USER_B => 1, ?USER_C => 1},
+                    await_tracking(Expected, Target),
+                    hold_connect_workers(Target, 1),
+                    [
+                        send_connect(Target, Sid, Sessions, 0)
+                     || Sid <- [<<"a1">>, <<"b1">>, <<"c1">>]
+                    ],
+                    ?assertEqual({error, overloaded}, await_result(<<"a1">>, 0)),
+                    ?assertEqual({error, overloaded}, await_result(<<"b1">>, 0)),
+                    State = get_state(Target),
+                    Dropped = maps:get(<<"a1">>, maps:get(sessions, State)),
+                    ?assertEqual(true, maps:get(pending_connect, Dropped)),
+                    ?assertNot(maps:is_key(<<"a1">>, maps:get(session_connect_pending, State))),
+                    assert_tracking(Expected, State),
+                    release_connect_workers(Target),
+                    ?assertMatch({ok, _, _}, await_result(<<"c1">>, 0)),
+                    send_connect(Target, <<"a1">>, Sessions, 1),
+                    send_connect(Target, <<"b1">>, Sessions, 1),
+                    ?assertMatch({ok, _, _}, await_result(<<"a1">>, 1)),
+                    ?assertMatch({ok, _, _}, await_result(<<"b1">>, 1)),
+                    await_tracking(Expected, Target),
+                    kill_session(<<"a1">>, Sessions),
+                    await_tracking(#{?USER_B => 1, ?USER_C => 1}, Target)
+                end)
+            end)
+        end
+    ).
+
+reconnect_of_connected_session_counts_once() ->
+    with_sessions(
+        [{<<"a1">>, ?USER_A}, {<<"b1">>, ?USER_B}],
+        fun(Sessions) ->
+            with_guild(base_state(Sessions), fun(Guild) ->
+                reconnect_all(Guild, Sessions, 1),
+                reconnect_all(Guild, Sessions, 2),
+                State = get_state(Guild),
+                assert_tracking(#{?USER_A => 1, ?USER_B => 1}, State),
+                [
+                    ?assertEqual(false, maps:get(pending_connect, Entry))
+                 || Entry <- maps:values(maps:get(sessions, State))
+                ],
+                kill_session(<<"b1">>, Sessions),
+                await_tracking(#{?USER_A => 1}, Guild)
+            end)
+        end
+    ).
+
+not_member_reconnect_releases_tracking() ->
+    with_sessions(
+        [{<<"a1">>, ?USER_A}, {<<"b1">>, ?USER_B}],
+        fun(Sessions) ->
+            with_guild(base_state(Sessions), fun(Guild) ->
+                sys:replace_state(Guild, fun(S) ->
+                    S#{data => guild_data_index:remove_member(?USER_B, maps:get(data, S))}
+                end),
+                send_connect(Guild, <<"b1">>, Sessions, 1),
+                ?assertEqual({error, not_member}, await_result(<<"b1">>, 1)),
+                State = get_state(Guild),
+                ?assertNot(maps:is_key(<<"b1">>, maps:get(sessions, State))),
+                assert_tracking(#{?USER_A => 1}, State)
+            end)
+        end
+    ).
+
+reconnecting_session_down_releases_tracking() ->
+    with_sessions(
+        [{<<"a1">>, ?USER_A}, {<<"b1">>, ?USER_B}],
+        fun(Sessions) ->
+            with_guild(base_state(Sessions), fun(Guild) ->
+                hold_connect_workers(Guild, 16),
+                send_connect(Guild, <<"b1">>, Sessions, 1),
+                Pending = maps:get(<<"b1">>, maps:get(sessions, get_state(Guild))),
+                ?assertEqual(true, maps:get(pending_connect, Pending)),
+                await_tracking(#{?USER_A => 1, ?USER_B => 1}, Guild),
+                kill_session(<<"b1">>, Sessions),
+                await_tracking(#{?USER_A => 1}, Guild)
+            end)
+        end
+    ).
+
+fresh_pending_down_keeps_reconnecting_owner_tracked() ->
+    with_sessions(
+        [{<<"a1">>, ?USER_A}, {<<"a2">>, ?USER_A}, {<<"c1">>, ?USER_C}],
+        fun(Sessions) ->
+            with_guild(base_state(without(<<"a1">>, Sessions)), fun(Guild) ->
+                hold_connect_workers(Guild, 16),
+                send_connect(Guild, <<"a2">>, Sessions, 1),
+                send_connect(Guild, <<"a1">>, Sessions, 1),
+                kill_session(<<"a1">>, Sessions),
+                await_tracking(#{?USER_A => 1, ?USER_C => 1}, Guild),
+                release_connect_workers(Guild),
+                ?assertMatch({ok, _, _}, await_result(<<"a2">>, 1)),
+                await_tracking(#{?USER_A => 1, ?USER_C => 1}, Guild),
+                kill_session(<<"a2">>, Sessions),
+                await_tracking(#{?USER_C => 1}, Guild)
+            end)
+        end
+    ).
+
+unavailable_removal_of_fresh_pending_keeps_owner_tracked() ->
+    with_sessions(
+        [{<<"a1">>, ?USER_A}, {<<"a2">>, ?USER_A}],
+        fun(Sessions) ->
+            with_guild(base_state(without(<<"a1">>, Sessions)), fun(Guild) ->
+                hold_connect_workers(Guild, 16),
+                send_connect(Guild, <<"a1">>, Sessions, 1),
+                sys:replace_state(Guild, fun(S) ->
+                    guild_sessions:remove_session(<<"a1">>, S)
+                end),
+                State = get_state(Guild),
+                ?assertNot(maps:is_key(<<"a1">>, maps:get(sessions, State))),
+                assert_tracking(#{?USER_A => 1}, State)
+            end)
+        end
+    ).
+
+fresh_pending_sibling_keeps_mutual_members() ->
+    with_sessions(
+        [{<<"a1">>, ?USER_A}, {<<"a2">>, ?USER_A}, {<<"c1">>, ?USER_C}],
+        fun(Sessions) ->
+            with_guild(base_state(without(<<"a1">>, Sessions)), fun(Guild) ->
+                send_connect(Guild, <<"a2">>, Sessions, 1),
+                ?assertMatch({ok, _, _}, await_result(<<"a2">>, 1)),
+                hold_connect_workers(Guild, 16),
+                send_connect(Guild, <<"a1">>, Sessions, 1),
+                State = get_state(Guild),
+                SessionMap = maps:get(<<"a2">>, maps:get(sessions, State)),
+                ?assertEqual(
+                    #{?CHANNEL_ID => true}, maps:get(viewable_channels, SessionMap)
+                ),
+                ?assertEqual(
+                    true,
+                    maps:get(pending_connect, maps:get(<<"a1">>, maps:get(sessions, State)))
+                ),
+                ?assertEqual(
+                    #{?CHANNEL_ID => true},
+                    guild_visibility_channels:get_cached_viewable_channel_map(?USER_A, State)
+                ),
+                ?assertEqual(
+                    [?USER_C],
+                    guild_subscription_mutual_channels:filter_member_ids(
+                        ?USER_A, [?USER_C], State
+                    )
+                )
+            end)
+        end
+    ).
+
+await_tracking(Expected, Guild) ->
+    await_tracking(Expected, Guild, 100).
+
+await_tracking(Expected, Guild, 0) ->
+    assert_tracking(Expected, get_state(Guild));
+await_tracking(Expected, Guild, Tries) ->
+    case tracking(get_state(Guild)) =:= expected_tracking(Expected) of
+        true ->
+            ok;
+        false ->
+            timer:sleep(20),
+            await_tracking(Expected, Guild, Tries - 1)
+    end.
+
+assert_tracking(Expected, State) ->
+    ?assertEqual(expected_tracking(Expected), tracking(State)).
+
+expected_tracking(Expected) ->
+    {Expected, lists:sort(maps:keys(Expected)), Expected}.
+
+tracking(State) ->
+    {
+        maps:get(user_session_counts, State),
+        lists:sort(sets:to_list(maps:get(connected_user_ids, State))),
+        maps:get(presence_subscriptions, State)
+    }.
+
+with_sessions(Specs, Fun) ->
+    Parent = self(),
+    Sessions = [
+        {Sid, UserId, spawn(fun() -> session_loop(Parent, Sid) end)}
+     || {Sid, UserId} <- Specs
+    ],
+    try
+        Fun(Sessions)
+    after
+        [exit(Pid, kill) || {_, _, Pid} <- Sessions],
+        flush()
+    end.
+
+session_loop(Parent, Sid) ->
+    receive
+        {guild_connect_result, _GuildId, Attempt, Reply} ->
+            Parent ! {connect_result, Sid, Attempt, Reply},
+            session_loop(Parent, Sid);
+        _ ->
+            session_loop(Parent, Sid)
+    end.
+
+without(Sid, Sessions) ->
+    lists:keydelete(Sid, 1, Sessions).
+
+with_guild(State, Fun) ->
+    {ok, Pid} = gen_server:start(guild, State, []),
+    try
+        Fun(Pid)
+    after
+        catch gen_server:call(Pid, {terminate}, 5000)
+    end.
+
+get_state(Guild) ->
+    gen_server:call(Guild, {get_sessions}, 10000).
+
+hold_connect_workers(Guild, MaxQueue) ->
+    sys:replace_state(Guild, fun(S) ->
+        S#{session_connect_inflight => 8, session_connect_max_queue => MaxQueue}
+    end),
+    ok.
+
+release_connect_workers(Guild) ->
+    sys:replace_state(Guild, fun(S) ->
+        guild_connect_async:maybe_start_session_connect_workers(
+            S#{session_connect_inflight => 0, session_connect_max_queue => 16}
+        )
+    end),
+    ok.
+
+reconnect_all(Guild, Sessions, Attempt) ->
+    [send_connect(Guild, Sid, Sessions, Attempt) || {Sid, _, _} <- Sessions],
+    [?assertMatch({ok, _, _}, await_result(Sid, Attempt)) || {Sid, _, _} <- Sessions],
+    ok.
+
+send_connect(Guild, Sid, Sessions, Attempt) ->
+    {Sid, UserId, Pid} = lists:keyfind(Sid, 1, Sessions),
+    Request = #{
+        session_id => Sid,
+        user_id => UserId,
+        session_pid => Pid,
+        bot => false,
+        is_staff => false,
+        initial_guild_id => ?GUILD_ID,
+        active_guilds => sets:from_list([?GUILD_ID])
+    },
+    gen_server:cast(
+        Guild,
+        {session_connect_async, #{
+            guild_id => ?GUILD_ID, attempt => Attempt, request => Request
+        }}
+    ).
+
+await_result(Sid, Attempt) ->
+    receive
+        {connect_result, Sid, Attempt, Reply} -> Reply
+    after 10000 ->
+        error({no_connect_result, Sid, Attempt})
+    end.
+
+kill_session(Sid, Sessions) ->
+    {Sid, _UserId, Pid} = lists:keyfind(Sid, 1, Sessions),
+    Ref = monitor(process, Pid),
+    exit(Pid, kill),
+    receive
+        {'DOWN', Ref, process, Pid, _} -> ok
+    end.
+
+flush() ->
+    receive
+        {connect_result, _, _, _} -> flush()
+    after 0 ->
+        ok
+    end.
+
+transferred_state(Exported) ->
+    guild_manager_shard_lifecycle:normalize_transferred_guild_state(?GUILD_ID, Exported).
+
+base_state(Sessions) ->
+    Members = [member(UserId) || UserId <- [?USER_A, ?USER_B, ?USER_C]],
+    #{
+        id => ?GUILD_ID,
+        member_count => length(Members),
+        sessions => maps:from_list([
+            {Sid, #{
+                session_id => Sid,
+                user_id => UserId,
+                pid => Pid,
+                active_guilds => sets:from_list([?GUILD_ID])
+            }}
+         || {Sid, UserId, Pid} <- Sessions
+        ]),
+        data => #{
+            <<"guild">> => #{
+                <<"id">> => ?GUILD_ID,
+                <<"owner_id">> => ?USER_A,
+                <<"features">> => [],
+                <<"member_count">> => length(Members)
+            },
+            <<"roles">> => [
+                #{
+                    <<"id">> => ?GUILD_ID,
+                    <<"name">> => <<"everyone">>,
+                    <<"permissions">> =>
+                        constants:view_channel_permission() bor
+                        constants:view_channel_members_permission(),
+                    <<"hoist">> => false,
+                    <<"position">> => 0
+                }
+            ],
+            <<"channels">> => [
+                #{
+                    <<"id">> => ?CHANNEL_ID,
+                    <<"name">> => <<"general">>,
+                    <<"type">> => 0,
+                    <<"permission_overwrites">> => []
+                }
+            ],
+            <<"members">> => Members
+        }
+    }.
+
+member(UserId) ->
+    Name = integer_to_binary(UserId),
+    #{
+        <<"user">> => #{
+            <<"id">> => UserId,
+            <<"username">> => <<"u", Name/binary>>,
+            <<"global_name">> => <<"U", Name/binary>>,
+            <<"bot">> => false
+        },
+        <<"nick">> => null,
+        <<"roles">> => [],
+        <<"joined_at">> => <<"2026-01-01T00:00:00.000000+00:00">>
+    }.