fix(gateway): bound relay worker mailboxes without reordering (#2309)

fluxer_gateway/src/gateway/gateway_dispatch_relay.erl

@@ -166,6 +166,20 @@ init({worker, Index}) ->
     {ok, #{role => worker, index => Index, delivered => 0}}.
 
 -spec handle_call(term(), gen_server:from(), state()) -> {reply, term(), state()}.
+handle_call(
+    {deliver, SessionPid, Event, Payload},
+    _From,
+    #{role := worker, delivered := Delivered} = State
+) ->
+    dispatch_direct(SessionPid, Event, Payload),
+    {reply, ok, State#{delivered := Delivered + 1}};
+handle_call(
+    {deliver_many, SessionPids, Event, Payload},
+    _From,
+    #{role := worker, delivered := Delivered} = State
+) when is_list(SessionPids) ->
+    deliver_many_direct(SessionPids, Event, Payload),
+    {reply, ok, State#{delivered := Delivered + length(SessionPids)}};
 handle_call(diagnostic_info, _From, State) ->
     {reply, diagnostic_info(), State};
 handle_call(_Request, _From, State) ->
@@ -182,12 +196,15 @@ handle_cast(
     {deliver_many, SessionPids, Event, Payload},
     #{role := worker, delivered := Delivered} = State
 ) when is_list(SessionPids) ->
-    Grouped = group_by_node(SessionPids),
+    deliver_many_direct(SessionPids, Event, Payload),
-    dispatch_grouped(Grouped, Event, Payload, fun dispatch_direct/3),
     {noreply, State#{delivered := Delivered + length(SessionPids)}};
 handle_cast(_Msg, State) ->
     {noreply, State}.
 
+-spec deliver_many_direct([pid()], term(), term()) -> ok.
+deliver_many_direct(SessionPids, Event, Payload) ->
+    dispatch_grouped(group_by_node(SessionPids), Event, Payload, fun dispatch_direct/3).
+
 -spec handle_info(term(), state()) -> {noreply, state()}.
 handle_info({'EXIT', Pid, Reason}, #{role := coordinator, workers := Workers} = State) when
     is_pid(Pid)

fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl

@@ -18,6 +18,7 @@
 ]).
 
 -define(STATE_KEY, {gateway_dispatch_relay, state}).
+-define(SYNC_TIMEOUT_MS, 5000).
 
 -spec relay_or_direct_many([pid()], atom(), term()) -> ok.
 relay_or_direct_many(SessionPids, Event, Payload) ->
@@ -68,7 +69,7 @@ deliver_shard_buckets(Index, Count, Buckets, Event, Payload, Workers) ->
 
 -spec deliver_shard(pos_integer(), [pid()], atom(), term(), tuple()) -> ok.
 deliver_shard(Index, Pids, Event, Payload, Workers) ->
-    gen_server:cast(element(Index, Workers), {deliver_many, Pids, Event, Payload}).
+    enqueue(element(Index, Workers), {deliver_many, Pids, Event, Payload}).
 
 -spec relay_or_direct(pid(), atom(), term()) -> ok.
 relay_or_direct(SessionPid, Event, Payload) ->
@@ -76,9 +77,33 @@ relay_or_direct(SessionPid, Event, Payload) ->
         undefined ->
             gateway_dispatch_relay:dispatch_direct(SessionPid, Event, Payload);
         Worker ->
-            gen_server:cast(Worker, {deliver, SessionPid, Event, Payload})
+            enqueue(Worker, {deliver, SessionPid, Event, Payload})
     end.
 
+-spec enqueue(pid(), term()) -> ok.
+enqueue(Worker, Msg) ->
+    case is_over_max_queue(Worker) of
+        true -> enqueue_sync(Worker, Msg);
+        false -> gen_server:cast(Worker, Msg)
+    end.
+
+-spec enqueue_sync(pid(), term()) -> ok.
+enqueue_sync(Worker, Msg) ->
+    try gen_server:call(Worker, Msg, ?SYNC_TIMEOUT_MS) of
+        _ -> ok
+    catch
+        exit:_Reason -> ok
+    end.
+
+-spec is_over_max_queue(pid()) -> boolean().
+is_over_max_queue(Worker) ->
+    MaxQueue = max_queue(),
+    MaxQueue > 0 andalso message_queue_len(Worker) >= MaxQueue.
+
+-spec max_queue() -> non_neg_integer().
+max_queue() ->
+    gateway_rollout_config:gateway_dispatch_relay_max_queue().
+
 -spec current_workers() -> [pid()].
 current_workers() ->
     tuple_to_list(current_workers_tuple()).

fluxer_gateway/src/telemetry/process_health_watchdog.erl

@@ -77,7 +77,15 @@ run_watchdog_check(#{history := History}) ->
 
 -spec collect_monitored_pids() -> [{pid(), binary()}].
 collect_monitored_pids() ->
-    guild_pids() ++ singleton_pids().
+    guild_pids() ++ singleton_pids() ++ dispatch_relay_worker_pids().
+
+-spec dispatch_relay_worker_pids() -> [{pid(), binary()}].
+dispatch_relay_worker_pids() ->
+    Workers = gateway_dispatch_relay_batch:current_workers(),
+    [
+        {Pid, iolist_to_binary(["dispatch_relay_worker:", integer_to_list(Index)])}
+     || {Index, Pid} <- lists:enumerate(0, Workers), is_pid(Pid)
+    ].
 
 -spec guild_pids() -> [{pid(), binary()}].
 guild_pids() ->
@@ -286,6 +294,19 @@ prune_dead_removes_absent_pids_test() ->
 resolve_singleton_missing_test() ->
     ?assertEqual(false, resolve_singleton(nonexistent_process_xyz_test)).
 
+dispatch_relay_workers_are_monitored_test() ->
+    Key = {gateway_dispatch_relay, state},
+    Previous = persistent_term:get(Key, undefined),
+    Worker = self(),
+    persistent_term:put(Key, #{workers => {Worker}, shard_count => 1}),
+    try
+        Expected = {Worker, <<"dispatch_relay_worker:0">>},
+        ?assertEqual([Expected], dispatch_relay_worker_pids()),
+        ?assert(lists:member(Expected, collect_monitored_pids()))
+    after
+        restore_relay_state_test_term(Key, Previous)
+    end.
+
 check_pid_dead_process_test() ->
     Pid = spawn(fun() -> ok end),
     ok = gateway_retry_timer:wait(50),
@@ -314,4 +335,10 @@ safe_run_watchdog_check_survives_bad_state_test() ->
     BadState = maps:remove(history, #{history => #{}}),
     ?assertEqual(BadState, safe_run_watchdog_check(eqwalizer:dynamic_cast(BadState))).
 
+restore_relay_state_test_term(Key, undefined) ->
+    _ = persistent_term:erase(Key),
+    ok;
+restore_relay_state_test_term(Key, Previous) ->
+    persistent_term:put(Key, Previous).
+
 -endif.

fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl

@@ -0,0 +1,178 @@
+%% SPDX-License-Identifier: AGPL-3.0-or-later
+
+-module(gateway_dispatch_relay_bound_tests).
+-typing([eqwalizer]).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(STATE_KEY, {gateway_dispatch_relay, state}).
+-define(CONFIG_KEY, gateway_rollout_config).
+-define(BOUND, 8).
+-define(FILL, 6).
+-define(EVENT_COUNT, 6).
+
+dispatch_is_bounded_and_stays_ordered_test_() ->
+    {timeout, 30, fun dispatch_is_bounded_and_stays_ordered/0}.
+
+dispatch_many_is_bounded_and_stays_ordered_test_() ->
+    {timeout, 30, fun dispatch_many_is_bounded_and_stays_ordered/0}.
+
+max_queue_config_governs_the_bound_test_() ->
+    {timeout, 30, fun max_queue_config_governs_the_bound/0}.
+
+max_queue_zero_disables_the_bound_test_() ->
+    {timeout, 30, fun max_queue_zero_disables_the_bound/0}.
+
+dispatch_is_bounded_and_stays_ordered() ->
+    assert_bounded_and_ordered(fun send_dispatch/2).
+
+dispatch_many_is_bounded_and_stays_ordered() ->
+    assert_bounded_and_ordered(fun send_dispatch_many/2).
+
+send_dispatch(SessionPid, N) ->
+    gateway_dispatch_relay:dispatch(SessionPid, relay_bound_event, #{<<"n">> => N}).
+
+send_dispatch_many(SessionPid, N) ->
+    gateway_dispatch_relay:dispatch_many([SessionPid], relay_bound_event, #{<<"n">> => N}).
+
+max_queue_config_governs_the_bound() ->
+    ?assertEqual(blocked, producer_status_with_bound(?BOUND, ?BOUND)),
+    ?assertEqual(finished, producer_status_with_bound(?BOUND * 8, ?BOUND)).
+
+max_queue_zero_disables_the_bound() ->
+    ?assertEqual(finished, producer_status_with_bound(0, ?BOUND)).
+
+assert_bounded_and_ordered(SendFun) ->
+    {Status, QueueLen, Observed} = run_over_bound(SendFun),
+    ?assertEqual(lists:seq(1, ?EVENT_COUNT), Observed),
+    ?assertEqual(blocked, Status),
+    ?assert(QueueLen =< ?BOUND + 1).
+
+run_over_bound(SendFun) ->
+    with_max_queue(?BOUND, fun() ->
+        with_worker(fun(Worker) ->
+            Ref = make_ref(),
+            Session = spawn_session(self(), Ref),
+            ok = sys:suspend(Worker),
+            fill_queue(Worker, ?FILL),
+            Producer = spawn_producer(SendFun, Session),
+            Status = producer_status(Producer, 500),
+            QueueLen = gateway_dispatch_relay_batch:message_queue_len(Worker),
+            ok = sys:resume(Worker),
+            ok = await_producer(Producer, Status),
+            Observed = collect_observed(Ref, ?EVENT_COUNT, []),
+            Session ! stop,
+            {Status, QueueLen, Observed}
+        end)
+    end).
+
+producer_status_with_bound(MaxQueue, Fill) ->
+    with_max_queue(MaxQueue, fun() ->
+        with_worker(fun(Worker) ->
+            Ref = make_ref(),
+            Session = spawn_session(self(), Ref),
+            ok = sys:suspend(Worker),
+            fill_queue(Worker, Fill),
+            Producer = spawn_producer(fun send_dispatch/2, Session),
+            Status = producer_status(Producer, 500),
+            ok = sys:resume(Worker),
+            ok = await_producer(Producer, Status),
+            ?assertEqual(lists:seq(1, ?EVENT_COUNT), collect_observed(Ref, ?EVENT_COUNT, [])),
+            Session ! stop,
+            Status
+        end)
+    end).
+
+spawn_producer(SendFun, Session) ->
+    Parent = self(),
+    spawn_link(fun() ->
+        lists:foreach(fun(N) -> ok = SendFun(Session, N) end, lists:seq(1, ?EVENT_COUNT)),
+        Parent ! {producer_finished, self()}
+    end).
+
+await_producer(_Producer, finished) ->
+    ok;
+await_producer(Producer, blocked) ->
+    ?assertEqual(finished, producer_status(Producer, 20000)),
+    ok.
+
+producer_status(Producer, Timeout) ->
+    receive
+        {producer_finished, Producer} -> finished
+    after Timeout ->
+        blocked
+    end.
+
+fill_queue(_Worker, 0) ->
+    ok;
+fill_queue(Worker, Remaining) ->
+    Worker ! relay_bound_filler,
+    fill_queue(Worker, Remaining - 1).
+
+spawn_session(Parent, Ref) ->
+    spawn_link(fun() -> session_loop(Parent, Ref) end).
+
+session_loop(Parent, Ref) ->
+    receive
+        {'$gen_cast', {dispatch, relay_bound_event, #{<<"n">> := N}}} ->
+            Parent ! {relay_bound_received, Ref, N},
+            session_loop(Parent, Ref);
+        stop ->
+            ok
+    after 30000 ->
+        ok
+    end.
+
+collect_observed(_Ref, 0, Acc) ->
+    lists:reverse(Acc);
+collect_observed(Ref, Remaining, Acc) ->
+    receive
+        {relay_bound_received, Ref, N} -> collect_observed(Ref, Remaining - 1, [N | Acc])
+    after 10000 ->
+        ?assert(false, {relay_bound_timeout, Remaining, lists:reverse(Acc)})
+    end.
+
+with_worker(Fun) ->
+    Worker = gateway_dispatch_relay_batch:start_worker(0),
+    try
+        with_relay_workers([Worker], fun() -> Fun(Worker) end)
+    after
+        stop_worker(Worker)
+    end.
+
+with_relay_workers(Workers, Fun) ->
+    Previous = persistent_term:get(?STATE_KEY, undefined),
+    persistent_term:put(?STATE_KEY, #{
+        workers => list_to_tuple(Workers),
+        shard_count => length(Workers)
+    }),
+    try
+        Fun()
+    after
+        restore_term(?STATE_KEY, Previous)
+    end.
+
+with_max_queue(MaxQueue, Fun) ->
+    Previous = persistent_term:get(?CONFIG_KEY, undefined),
+    Config = gateway_rollout_config:get(),
+    persistent_term:put(?CONFIG_KEY, Config#{
+        <<"gateway_dispatch_relay_max_queue">> => MaxQueue
+    }),
+    try
+        Fun()
+    after
+        restore_term(?CONFIG_KEY, Previous)
+    end.
+
+restore_term(Key, undefined) ->
+    _ = persistent_term:erase(Key),
+    ok;
+restore_term(Key, Previous) ->
+    persistent_term:put(Key, Previous).
+
+stop_worker(Pid) ->
+    try gen_server:stop(Pid, normal, 5000) of
+        _ -> ok
+    catch
+        exit:_ -> ok
+    end.

fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl

@@ -38,7 +38,7 @@ ordered_delivery(Worker, SendFun) ->
     SessionPid = spawn_link(fun() -> session_loop(Parent, Ref) end),
     MaxQueue = gateway_rollout_config:gateway_dispatch_relay_max_queue(),
     ok = sys:suspend(Worker),
-    fill_queue(Worker, MaxQueue - 1),
+    fill_queue(Worker, MaxQueue - ?EVENT_COUNT),
     lists:foreach(fun(N) -> ok = SendFun(SessionPid, N) end, lists:seq(1, ?EVENT_COUNT)),
     ?assert(gateway_dispatch_relay_batch:message_queue_len(Worker) >= MaxQueue),
     ok = sys:resume(Worker),