fix(gateway): bound relay dispatch without per-event probes (#2312)

fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl

@@ -12,6 +12,7 @@
     current_workers_tuple_normalized/0,
     normalize_workers_tuple/1,
     message_queue_len/1,
+    max_queue/0,
     start_workers/1,
     start_worker/1,
     worker_index/3
@@ -19,6 +20,7 @@
 
 -define(STATE_KEY, {gateway_dispatch_relay, state}).
 -define(SYNC_TIMEOUT_MS, 5000).
+-define(SLOT_KEY(Worker), {?MODULE, Worker}).
 
 -spec relay_or_direct_many([pid()], atom(), term()) -> ok.
 relay_or_direct_many(SessionPids, Event, Payload) ->
@@ -39,7 +41,8 @@ relay_or_direct_many(SessionPids, Event, Payload) ->
 -spec relay_many_to_shards([pid()], atom(), term(), tuple(), pos_integer()) -> ok.
 relay_many_to_shards(SessionPids, Event, Payload, Workers, Count) ->
     ShardBuckets = build_shard_buckets(SessionPids, Count),
-    deliver_shard_buckets(1, Count, ShardBuckets, Event, Payload, Workers).
+    Deferred = deliver_shard_buckets(1, Count, ShardBuckets, Event, Payload, Workers, []),
+    deliver_deferred_shards(Deferred, Event, Payload).
 
 -spec build_shard_buckets([pid()], pos_integer()) -> tuple().
 build_shard_buckets(SessionPids, Count) ->
@@ -56,20 +59,35 @@ build_shard_buckets(SessionPids, Count) ->
     ).
 
 -spec deliver_shard_buckets(
-    pos_integer(), pos_integer(), tuple(), atom(), term(), tuple()
-) -> ok.
-deliver_shard_buckets(Index, Count, _Buckets, _Event, _Payload, _Workers) when Index > Count ->
-    ok;
-deliver_shard_buckets(Index, Count, Buckets, Event, Payload, Workers) ->
-    case element(Index, Buckets) of
-        [] -> ok;
-        Pids -> deliver_shard(Index, Pids, Event, Payload, Workers)
-    end,
-    deliver_shard_buckets(Index + 1, Count, Buckets, Event, Payload, Workers).
+    pos_integer(), pos_integer(), tuple(), atom(), term(), tuple(), [{pid(), [pid()]}]
+) -> [{pid(), [pid()]}].
+deliver_shard_buckets(Index, Count, _Buckets, _Event, _Payload, _Workers, Deferred) when
+    Index > Count
+->
+    Deferred;
+deliver_shard_buckets(Index, Count, Buckets, Event, Payload, Workers, Deferred) ->
+    Next =
+        case element(Index, Buckets) of
+            [] -> Deferred;
+            Pids -> deliver_shard(Index, Pids, Event, Payload, Workers, Deferred)
+        end,
+    deliver_shard_buckets(Index + 1, Count, Buckets, Event, Payload, Workers, Next).
 
--spec deliver_shard(pos_integer(), [pid()], atom(), term(), tuple()) -> ok.
-deliver_shard(Index, Pids, Event, Payload, Workers) ->
-    enqueue(element(Index, Workers), {deliver_many, Pids, Event, Payload}).
+-spec deliver_shard(pos_integer(), [pid()], atom(), term(), tuple(), [{pid(), [pid()]}]) ->
+    [{pid(), [pid()]}].
+deliver_shard(Index, Pids, Event, Payload, Workers, Deferred) ->
+    Worker = element(Index, Workers),
+    case enqueue_async(Worker, {deliver_many, Pids, Event, Payload}) of
+        ok -> Deferred;
+        full -> [{Worker, Pids} | Deferred]
+    end.
+
+-spec deliver_deferred_shards([{pid(), [pid()]}], atom(), term()) -> ok.
+deliver_deferred_shards([], _Event, _Payload) ->
+    ok;
+deliver_deferred_shards([{Worker, Pids} | Rest], Event, Payload) ->
+    enqueue_sync(Worker, {deliver_many, Pids, Event, Payload}),
+    deliver_deferred_shards(Rest, Event, Payload).
 
 -spec relay_or_direct(pid(), atom(), term()) -> ok.
 relay_or_direct(SessionPid, Event, Payload) ->
@@ -82,9 +100,37 @@ relay_or_direct(SessionPid, Event, Payload) ->
 
 -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)
+    case enqueue_async(Worker, Msg) of
+        ok -> ok;
+        full -> enqueue_sync(Worker, Msg)
+    end.
+
+-spec enqueue_async(pid(), term()) -> ok | full.
+enqueue_async(Worker, Msg) ->
+    case claim_queue_slot(Worker, max_queue()) of
+        ok -> gen_server:cast(Worker, Msg);
+        full -> full
+    end.
+
+-spec claim_queue_slot(pid(), pos_integer()) -> ok | full.
+claim_queue_slot(Worker, MaxQueue) ->
+    case erlang:get(?SLOT_KEY(Worker)) of
+        Claimed when is_integer(Claimed), Claimed < MaxQueue ->
+            _ = erlang:put(?SLOT_KEY(Worker), Claimed + 1),
+            ok;
+        _ ->
+            sample_queue_slot(Worker, MaxQueue)
+    end.
+
+-spec sample_queue_slot(pid(), pos_integer()) -> ok | full.
+sample_queue_slot(Worker, MaxQueue) ->
+    case message_queue_len(Worker) of
+        Sampled when Sampled < MaxQueue ->
+            _ = erlang:put(?SLOT_KEY(Worker), Sampled + 1),
+            ok;
+        _ ->
+            _ = erlang:erase(?SLOT_KEY(Worker)),
+            full
     end.
 
 -spec enqueue_sync(pid(), term()) -> ok.
@@ -95,14 +141,13 @@ enqueue_sync(Worker, Msg) ->
         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().
+-spec max_queue() -> pos_integer().
 max_queue() ->
-    gateway_rollout_config:gateway_dispatch_relay_max_queue().
+    Ceiling = process_health_watchdog:kill_threshold(),
+    case gateway_rollout_config:gateway_dispatch_relay_max_queue() of
+        Configured when is_integer(Configured), Configured > 0 -> min(Configured, Ceiling);
+        _ -> Ceiling
+    end.
 
 -spec current_workers() -> [pid()].
 current_workers() ->

fluxer_gateway/src/telemetry/process_health_watchdog.erl

@@ -4,7 +4,7 @@
 -typing([eqwalizer]).
 -behaviour(gen_server).
 
--export([start_link/0]).
+-export([start_link/0, kill_threshold/0]).
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
 
 -define(CHECK_INTERVAL_MS, 10_000).
@@ -17,6 +17,10 @@
 -type queue_history() :: #{pid() => [non_neg_integer()]}.
 -type state() :: #{history := queue_history()}.
 
+-spec kill_threshold() -> pos_integer().
+kill_threshold() ->
+    ?KILL_THRESHOLD.
+
 -spec start_link() -> {ok, pid()} | {error, term()}.
 start_link() ->
     case gen_server:start_link({local, ?MODULE}, ?MODULE, [], []) of

fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl

@@ -10,6 +10,11 @@
 -define(BOUND, 8).
 -define(FILL, 6).
 -define(EVENT_COUNT, 6).
+-define(PROBE_BOUND, 1000).
+-define(PROBE_EVENT_COUNT, 200).
+-define(PROBE_BUDGET, 2).
+-define(SHARD_COUNT, 2).
+-define(FAST_SHARD_TIMEOUT_MS, 1500).
 
 dispatch_is_bounded_and_stays_ordered_test_() ->
     {timeout, 30, fun dispatch_is_bounded_and_stays_ordered/0}.
@@ -20,8 +25,17 @@ dispatch_many_is_bounded_and_stays_ordered_test_() ->
 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}.
+max_queue_zero_keeps_the_bound_test_() ->
+    {timeout, 60, fun max_queue_zero_keeps_the_bound/0}.
+
+max_queue_is_capped_by_the_watchdog_kill_threshold_test_() ->
+    {timeout, 30, fun max_queue_is_capped_by_the_watchdog_kill_threshold/0}.
+
+dispatch_does_not_probe_the_worker_per_event_test_() ->
+    {timeout, 30, fun dispatch_does_not_probe_the_worker_per_event/0}.
+
+saturated_shard_does_not_delay_other_shards_test_() ->
+    {timeout, 60, fun saturated_shard_does_not_delay_other_shards/0}.
 
 dispatch_is_bounded_and_stays_ordered() ->
     assert_bounded_and_ordered(fun send_dispatch/2).
@@ -39,8 +53,57 @@ 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)).
+max_queue_zero_keeps_the_bound() ->
+    Ceiling = process_health_watchdog:kill_threshold(),
+    with_max_queue(0, fun() ->
+        ?assertEqual(Ceiling, gateway_dispatch_relay_batch:max_queue())
+    end),
+    ?assertEqual(blocked, producer_status_with_bound(0, Ceiling)).
+
+max_queue_is_capped_by_the_watchdog_kill_threshold() ->
+    Ceiling = process_health_watchdog:kill_threshold(),
+    with_max_queue(Ceiling * 4, fun() ->
+        ?assertEqual(Ceiling, gateway_dispatch_relay_batch:max_queue())
+    end),
+    with_max_queue(Ceiling - 1, fun() ->
+        ?assertEqual(Ceiling - 1, gateway_dispatch_relay_batch:max_queue())
+    end).
+
+dispatch_does_not_probe_the_worker_per_event() ->
+    with_max_queue(?PROBE_BOUND, fun() ->
+        with_worker(fun(Worker) ->
+            Ref = make_ref(),
+            Session = spawn_session(self(), Ref),
+            Producer = spawn_gated_producer(Session, ?PROBE_EVENT_COUNT),
+            Probes = count_probes(Worker, Producer),
+            ?assertEqual(
+                lists:seq(1, ?PROBE_EVENT_COUNT),
+                collect_observed(Ref, ?PROBE_EVENT_COUNT, [])
+            ),
+            Session ! stop,
+            ?assert(Probes =< ?PROBE_BUDGET)
+        end)
+    end).
+
+saturated_shard_does_not_delay_other_shards() ->
+    with_max_queue(?BOUND, fun() ->
+        with_workers(?SHARD_COUNT, fun([Blocked, _Free]) ->
+            SlowRef = make_ref(),
+            FastRef = make_ref(),
+            Slow = session_for_shard(1, self(), SlowRef),
+            Fast = session_for_shard(2, self(), FastRef),
+            ok = sys:suspend(Blocked),
+            fill_queue(Blocked, ?BOUND),
+            Producer = spawn_fanout_producer([Slow, Fast]),
+            ?assertEqual([1], collect_within(FastRef, ?FAST_SHARD_TIMEOUT_MS)),
+            ?assertEqual(blocked, producer_status(Producer, 100)),
+            ok = sys:resume(Blocked),
+            ?assertEqual([1], collect_observed(SlowRef, 1, [])),
+            ?assertEqual(finished, producer_status(Producer, 20000)),
+            Slow ! stop,
+            Fast ! stop
+        end)
+    end).
 
 assert_bounded_and_ordered(SendFun) ->
     {Status, QueueLen, Observed} = run_over_bound(SendFun),
@@ -90,6 +153,58 @@ spawn_producer(SendFun, Session) ->
         Parent ! {producer_finished, self()}
     end).
 
+spawn_gated_producer(Session, Count) ->
+    Parent = self(),
+    spawn_link(fun() ->
+        receive
+            start -> ok
+        end,
+        lists:foreach(fun(N) -> ok = send_dispatch(Session, N) end, lists:seq(1, Count)),
+        Parent ! {producer_finished, self()}
+    end).
+
+spawn_fanout_producer(Sessions) ->
+    Parent = self(),
+    spawn_link(fun() ->
+        ok = gateway_dispatch_relay:dispatch_many(Sessions, relay_bound_event, #{<<"n">> => 1}),
+        Parent ! {producer_finished, self()}
+    end).
+
+session_for_shard(Index, Parent, Ref) ->
+    Session = spawn_session(Parent, Ref),
+    case erlang:phash2(Session, ?SHARD_COUNT) + 1 of
+        Index -> Session;
+        _ -> session_for_shard(Index, Parent, Ref)
+    end.
+
+count_probes(Worker, Producer) ->
+    1 = erlang:trace(Producer, true, [call]),
+    _ = erlang:trace_pattern({erlang, process_info, 2}, true, [global]),
+    Producer ! start,
+    try
+        drain_probes(Worker, Producer, 0)
+    after
+        _ = erlang:trace_pattern({erlang, process_info, 2}, false, [global]),
+        untrace(Producer)
+    end.
+
+untrace(Producer) ->
+    try erlang:trace(Producer, false, [call]) of
+        _ -> ok
+    catch
+        error:badarg -> ok
+    end.
+
+drain_probes(Worker, Producer, Count) ->
+    receive
+        {trace, Producer, call, {erlang, process_info, [Worker, message_queue_len]}} ->
+            drain_probes(Worker, Producer, Count + 1);
+        {producer_finished, Producer} ->
+            Count
+    after 20000 ->
+        ?assert(false, {probe_count_timeout, Count})
+    end.
+
 await_producer(_Producer, finished) ->
     ok;
 await_producer(Producer, blocked) ->
@@ -123,6 +238,13 @@ session_loop(Parent, Ref) ->
         ok
     end.
 
+collect_within(Ref, Timeout) ->
+    receive
+        {relay_bound_received, Ref, N} -> [N]
+    after Timeout ->
+        []
+    end.
+
 collect_observed(_Ref, 0, Acc) ->
     lists:reverse(Acc);
 collect_observed(Ref, Remaining, Acc) ->
@@ -133,11 +255,17 @@ collect_observed(Ref, Remaining, Acc) ->
     end.
 
 with_worker(Fun) ->
-    Worker = gateway_dispatch_relay_batch:start_worker(0),
+    with_workers(1, fun([Worker]) -> Fun(Worker) end).
+
+with_workers(Count, Fun) ->
+    Workers = [
+        gateway_dispatch_relay_batch:start_worker(Index)
+     || Index <- lists:seq(0, Count - 1)
+    ],
     try
-        with_relay_workers([Worker], fun() -> Fun(Worker) end)
+        with_relay_workers(Workers, fun() -> Fun(Workers) end)
     after
-        stop_worker(Worker)
+        lists:foreach(fun stop_worker/1, Workers)
     end.
 
 with_relay_workers(Workers, Fun) ->

fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl

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