MB-67857: Avoid menelaus_cbauth_worker message queue growing

Because we synchronously notify cbauth instances of creds info, and
the creds info is generated outside the worker process, we can end up
with a queue of updates filling up the menelaus_cbauth_worker process.

The obvious solution would be to simply generate the creds info within
the worker process, such that we could instead flush all notify
messages each time we process one, avoiding the message queue backing
up. The issue with this is that it would mean duplicating the
management of creds info in each worker process, which is currently
handled in just menelaus_cbauth.

Instead, it is simpler just to flush the notify messages when we
process one, while retaining the info from the last message.
This avoids the message queue backing up, ensuring that each time we
notify cbauth, it is with the latest info that the process received.

Change-Id: I3c3b15dd7b81f005524a24d0614895a6e8fcb79f
Reviewed-on: https://review.couchbase.org/c/ns_server/+/231636
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Timofey Barmin <[email protected]>
Well-Formed: Restriction Checker
Tested-by: Peter Searby <[email protected]>

Author: Peter-Searby

src/menelaus_cbauth.erl

@@ -771,6 +771,49 @@ cbauth_stats_t() ->
                 end),
     [] = stats().
 
+cbauth_many_notify_t() ->
+    %% Wait for initial notify to be handled
+    meck:wait(1, menelaus_cbauth_worker, notify, ['_', '_'],
+              ?LONG_TIMEOUT),
+    meck:wait(1, json_rpc_connection, perform_call,
+              ['_', "AuthCacheSvc.UpdateDB", '_', '_'], ?LONG_TIMEOUT),
+    %% Ensure that the next perform_call doesn't immediately return
+    meck:expect(json_rpc_connection, perform_call,
+                fun (_, "AuthCacheSvc.UpdateDB", _, _) ->
+                        receive return_from_call -> {ok, true}
+                        after ?LONG_TIMEOUT -> error(timeout) end;
+                    (_Label, _Call, _EJsonArg, _Opts) ->
+                        {ok, true}
+                end),
+    %% Send three updates, ensuring three notify calls with different values:
+    %% - The first is to get the worker stuck in the above meck:expect
+    %% - The second is to provide a value, which we want to become stale
+    %% - The third will be the new value, which should get used, instead of the
+    %%   stale value, after the first update is processed
+    lists:foreach(
+      fun (N) ->
+              fake_ns_config:update_snapshot(rest, [{port, N}]),
+              meck:wait(N, menelaus_cbauth_worker, notify, ['_', '_'],
+                        ?LONG_TIMEOUT)
+      end, [2, 3, 4]),
+    %% Extract value from next update
+    meck:expect(json_rpc_connection, perform_call,
+                fun (_, "AuthCacheSvc.UpdateDB", {Info}, _) ->
+                        [{Node}] = proplists:get_value(nodes, Info, []),
+                        [undefined, Port] = proplists:get_value(ports, Node),
+                        %% Expect the last value, not the stale value
+                        ?assertEqual(4, Port),
+                        {ok, true};
+                    (_Label, _Call, _EJsonArg, _Opts) ->
+                        {ok, true}
+                end),
+    %% Release worker
+    WorkerPid = whereis(list_to_atom("menelaus_cbauth_worker-" ++ ?LABEL)),
+    WorkerPid ! return_from_call,
+    %% Wait for both the stuck and final call to return
+    meck:wait(3, json_rpc_connection, perform_call,
+              ['_', "AuthCacheSvc.UpdateDB", '_', '_'], ?LONG_TIMEOUT).
+
 trigger_notification(ns_node_disco_events) ->
     %% To avoid needing to trick ns_node_disco into recognising fake nodes, just
     %% make the node list different by removing the only node.
@@ -823,6 +866,7 @@ cbauth_test_() ->
     {foreach, fun setup_t/0, fun teardown_t/1,
      [{"cbauth init test", fun cbauth_init_t/0},
       {"cbauth sync test", fun cbauth_sync_t/0},
-      {"cbauth stats test", fun cbauth_stats_t/0}
+      {"cbauth stats test", fun cbauth_stats_t/0},
+      {"cbauth many notify test", fun cbauth_many_notify_t/0}
      | cbauth_notify_tests()]}.
 -endif.

src/menelaus_cbauth_worker.erl

@@ -82,17 +82,18 @@ handle_info(heartbeat, State = #state{label = Label, connection = Pid,
         ok ->
             {noreply, State}
     end;
-handle_info({notify, InfoHidden}, State = #state{label = Label,
-                                                 connection = Pid,
-                                                 version = Version}) ->
-    Info = ?UNHIDE(InfoHidden),
+handle_info({notify, PotentiallyStaleInfo},
+            State = #state{label = Label, connection = Pid,
+                           version = Version}) ->
+    LatestInfoHidden = receive_latest_notify(PotentiallyStaleInfo),
+    LatestInfo = ?UNHIDE(LatestInfoHidden),
     Method = case Version of
                  internal ->
                      "AuthCacheSvc.UpdateDB";
                  _ ->
                      "AuthCacheSvc.UpdateDBExt"
              end,
-    case invoke_no_return_method(Label, Method, Pid, Info) of
+    case invoke_no_return_method(Label, Method, Pid, LatestInfo) of
         error ->
             terminate_jsonrpc_connection(Label, Pid),
             misc:wait_for_process(Pid, infinity),
@@ -136,6 +137,23 @@ send_heartbeat(Label, Pid) ->
             ?log_debug("Skip heartbeat for label ~p", [Label])
     end.
 
+receive_latest_notify(Info) ->
+    receive_latest_notify(Info, 0).
+
+receive_latest_notify(Info, N) ->
+    %% Handle any remaining notify messages, to ensure that the mailbox is
+    %% cleared of redundant messages
+    receive
+        {notify, NewInfo} -> receive_latest_notify(NewInfo, N + 1)
+    after
+        0 ->
+            case N of
+                0 -> ok;
+                _ -> ?log_debug("Skipped ~p old notify entries", [N])
+            end,
+            Info
+    end.
+
 invoke_no_return_method(Label, Method, Pid, Info) ->
     case perform_call(Label, Method, Pid, {Info}, false) of
         {ok, Res} when Res =:= true orelse Res =:= null ->