From da3332e711bff0c4039b8379c852756350aafeb0 Mon Sep 17 00:00:00 2001 From: Hampus Date: Tue, 1 Sep 2026 01:38:36 +0200 Subject: [PATCH] fix(gateway): bound relay worker mailboxes without reordering (#2309) --- .../src/gateway/gateway_dispatch_relay.erl | 21 ++- .../gateway/gateway_dispatch_relay_batch.erl | 29 ++- .../src/telemetry/process_health_watchdog.erl | 29 ++- .../gateway_dispatch_relay_bound_tests.erl | 178 ++++++++++++++++++ .../gateway_dispatch_relay_ordering_tests.erl | 2 +- 5 files changed, 253 insertions(+), 6 deletions(-) create mode 100644 fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl diff --git a/fluxer_gateway/src/gateway/gateway_dispatch_relay.erl b/fluxer_gateway/src/gateway/gateway_dispatch_relay.erl index 19af2c526..013b0f576 100644 --- a/fluxer_gateway/src/gateway/gateway_dispatch_relay.erl +++ b/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), - dispatch_grouped(Grouped, Event, Payload, fun dispatch_direct/3), + deliver_many_direct(SessionPids, Event, Payload), {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) diff --git a/fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl b/fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl index aa65ce638..df1f6c6aa 100644 --- a/fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl +++ b/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()). diff --git a/fluxer_gateway/src/telemetry/process_health_watchdog.erl b/fluxer_gateway/src/telemetry/process_health_watchdog.erl index e1a946fa9..07879a594 100644 --- a/fluxer_gateway/src/telemetry/process_health_watchdog.erl +++ b/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. diff --git a/fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl b/fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl new file mode 100644 index 000000000..8e7ee29e6 --- /dev/null +++ b/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. diff --git a/fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl b/fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl index b9c607d1c..38ec87e76 100644 --- a/fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl +++ b/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),