Be more defensive against duplicate JSON keys when management stats are enabled

michaelklishin · michaelklishin · commit fa753a12fcb2 · 2026-03-05T19:45:37.000-08:00
`rabbit_queue_type:format/2` now conditionally emits policy fields only when management stats are disabled: the stats augmentation code path provides them. 1. `format/2` now conditionally emits policy fields based on whether the management plugin stats are enabled 2. `combine/2` acts as the next layer of protection against duplicate keys: it drops the `format/2`-originating keys that are duplicates 3. `prepare_for_encoding/1` is the new name for `format_nulls/1` that reflects its expanded scope: another deduplication pass References #15182.
diff --git a/deps/rabbit/src/rabbit_classic_queue.erl b/deps/rabbit/src/rabbit_classic_queue.erl
@@ -264,7 +264,7 @@ stat(Q) ->
                     {gen_server2, call, [stat, infinity]}).
 
 
-format(Q, _Ctx) when ?is_amqqueue(Q) ->
+format(Q, Ctx) when ?is_amqqueue(Q) ->
     State = case amqqueue:get_state(Q) of
                 live ->
                     running;
@@ -273,10 +273,18 @@ format(Q, _Ctx) when ?is_amqqueue(Q) ->
             end,
     [{type, rabbit_queue_type:short_alias_of(?MODULE)},
      {state, State},
-     {node, node(amqqueue:get_pid(Q))},
-     {policy, i(policy, Q)},
-     {operator_policy, i(operator_policy, Q)},
-     {effective_policy_definition, i(effective_policy_definition, Q)}].
+     {node, node(amqqueue:get_pid(Q))}
+     | format_policy_fields(Q, Ctx)].
+
+format_policy_fields(Q, Ctx) ->
+    case maps:get(management_stats_disabled, Ctx, true) of
+        true ->
+            [{policy, i(policy, Q)},
+             {operator_policy, i(operator_policy, Q)},
+             {effective_policy_definition, i(effective_policy_definition, Q)}];
+        false ->
+            []
+    end.
 
 -spec init(amqqueue:amqqueue()) -> {ok, state()}.
 init(Q) when ?amqqueue_is_classic(Q) ->
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -2154,11 +2154,19 @@ format(Q, Ctx) when ?is_amqqueue(Q) ->
      {node, LeaderNode},
      {members, Nodes},
      {leader, LeaderNode},
-     {online, Online},
-     {policy, i(policy, Q)},
-     {operator_policy, i(operator_policy, Q)},
-     {effective_policy_definition, i(effective_policy_definition, Q)},
-     {delivery_limit, i(delivery_limit, Q)}].
+     {online, Online}
+     | format_policy_fields(Q, Ctx)].
+
+format_policy_fields(Q, Ctx) ->
+    case maps:get(management_stats_disabled, Ctx, true) of
+        true ->
+            [{policy, i(policy, Q)},
+             {operator_policy, i(operator_policy, Q)},
+             {effective_policy_definition, i(effective_policy_definition, Q)},
+             {delivery_limit, i(delivery_limit, Q)}];
+        false ->
+            []
+    end.
 
 -spec quorum_messages(rabbit_amqqueue:name()) -> non_neg_integer().
 
diff --git a/deps/rabbit/src/rabbit_stream_queue.erl b/deps/rabbit/src/rabbit_stream_queue.erl
@@ -303,16 +303,22 @@ format(Q, Ctx) ->
              {leader, LeaderNode},
              {online, Online},
              {members, Nodes},
-             {node, node(Pid)},
-             {policy, i(policy, Q)},
-             {operator_policy, i(operator_policy, Q)},
-             {effective_policy_definition, i(effective_policy_definition, Q)}];
+             {node, node(Pid)}
+             | format_policy_fields(Q, Ctx)];
         _ ->
             [{type, rabbit_queue_type:short_alias_of(?MODULE)},
-             {state, down},
-             {policy, i(policy, Q)},
+             {state, down}
+             | format_policy_fields(Q, Ctx)]
+    end.
+
+format_policy_fields(Q, Ctx) ->
+    case maps:get(management_stats_disabled, Ctx, true) of
+        true ->
+            [{policy, i(policy, Q)},
              {operator_policy, i(operator_policy, Q)},
-             {effective_policy_definition, i(effective_policy_definition, Q)}]
+             {effective_policy_definition, i(effective_policy_definition, Q)}];
+        false ->
+            []
     end.
 
 consume(Q, #{mode := {simple_prefetch, 0}}, _)
diff --git a/deps/rabbit_common/src/rabbit_json.erl b/deps/rabbit_common/src/rabbit_json.erl
@@ -46,7 +46,7 @@ encode(Term) ->
 encode(Term, Opts) ->
     %% Fixup for JSON encoding
     %% * Transforms any Funs into strings
-    %% See rabbit_mgmt_format:format_nulls/1
+    %% See rabbit_mgmt_format:prepare_for_encoding/1
     F = fun
             (V) when is_function(V) ->
                 rabbit_data_coercion:to_binary(V);
diff --git a/deps/rabbitmq_management/src/rabbit_mgmt_db.erl b/deps/rabbitmq_management/src/rabbit_mgmt_db.erl
@@ -665,10 +665,13 @@ node_stats(Ranges, Objs, Interval) ->
      end || Obj <- Objs].
 
 combine(New, Old) ->
+    NewKeys = [K || {K, _} <- New],
     case pget(state, Old) of
-        unknown -> New ++ Old;
-        live    -> New ++ delete_keys([state, online], Old);
-        _       -> lists:keydelete(state, 1, New) ++ Old
+        unknown -> New ++ delete_keys(NewKeys, Old);
+        live    -> New ++ delete_keys([state, online | NewKeys], Old);
+        _       ->
+            NewKeysNoState = lists:delete(state, NewKeys),
+            lists:keydelete(state, 1, New) ++ delete_keys(NewKeysNoState, Old)
     end.
 
 delete_keys(Keys, List) ->
diff --git a/deps/rabbitmq_management/src/rabbit_mgmt_oauth_bootstrap.erl b/deps/rabbitmq_management/src/rabbit_mgmt_oauth_bootstrap.erl
@@ -103,7 +103,7 @@ validate_auth_mechanism({Type, <<"basic">>}, _AuthSettings) ->
 validate_auth_mechanism({_, _}, _AuthSettings) -> {error, unknown_auth_mechanism}.
 
 set_oauth_settings(AuthSettings) ->
-    JsonAuthSettings = rabbit_json:encode(rabbit_mgmt_format:format_nulls(AuthSettings)),
+    JsonAuthSettings = rabbit_json:encode(rabbit_mgmt_format:prepare_for_encoding(AuthSettings)),
     ["set_oauth_settings(", JsonAuthSettings, ");"].
 
 set_token_auth(AuthSettings, Req0) ->
diff --git a/deps/rabbitmq_management/src/rabbit_mgmt_util.erl b/deps/rabbitmq_management/src/rabbit_mgmt_util.erl
@@ -267,7 +267,7 @@ reply0(Facts, ReqData, Context) ->
             {<<"application">>, <<"bert">>, _} ->
                 {term_to_binary(Facts), ReqData1, Context};
             _ ->
-                {rabbit_json:encode(rabbit_mgmt_format:format_nulls(Facts)),
+                {rabbit_json:encode(rabbit_mgmt_format:prepare_for_encoding(Facts)),
                  ReqData1, Context}
         end
     catch exit:{json_encode, E} ->
diff --git a/deps/rabbitmq_management/src/rabbit_mgmt_wm_queue.erl b/deps/rabbitmq_management/src/rabbit_mgmt_wm_queue.erl
@@ -39,17 +39,19 @@ resource_exists(ReqData, Context) ->
 
 to_json(ReqData, Context) ->
     try
-        case rabbit_mgmt_util:disable_stats(ReqData) of
+        StatsDisabled = rabbit_mgmt_util:disable_stats(ReqData),
+        case StatsDisabled of
             false ->
                 [Q] = rabbit_mgmt_db:augment_queues(
-                        [queue(ReqData)], rabbit_mgmt_util:range_ceil(ReqData),
+                        [queue(ReqData, StatsDisabled)],
+                        rabbit_mgmt_util:range_ceil(ReqData),
                         full),
                 Payload = rabbit_mgmt_format:clean_consumer_details(
                             rabbit_mgmt_format:strip_pids(Q)),
                 rabbit_mgmt_util:reply(ensure_defaults(Payload), ReqData, Context);
             true ->
                 Q = case rabbit_mgmt_util:enable_queue_totals(ReqData) of
-                    false -> queue(ReqData);
+                    false -> queue(ReqData, StatsDisabled);
                     true  -> queue_with_totals(ReqData)
                 end,
                 rabbit_mgmt_util:reply(
@@ -108,15 +110,20 @@ ensure_defaults(Payload0) ->
     end.
 
 queue(ReqData) ->
+    queue(ReqData, rabbit_mgmt_util:disable_stats(ReqData)).
+
+queue(ReqData, StatsDisabled) ->
     case rabbit_mgmt_util:vhost(ReqData) of
         not_found -> not_found;
-        VHost     -> queue(VHost, rabbit_mgmt_util:id(queue, ReqData))
+        VHost     ->
+            Ctx = #{management_stats_disabled => StatsDisabled},
+            queue(VHost, rabbit_mgmt_util:id(queue, ReqData), Ctx)
     end.
 
-queue(VHost, QName) ->
+queue(VHost, QName, Ctx) ->
     Name = rabbit_misc:r(VHost, queue, QName),
     case rabbit_amqqueue:lookup(Name) of
-        {ok, Q}            -> rabbit_mgmt_format:queue(Q);
+        {ok, Q}            -> rabbit_mgmt_format:queue(Q, Ctx);
         {error, not_found} -> not_found
     end.
 
diff --git a/deps/rabbitmq_management/src/rabbit_mgmt_wm_queues.erl b/deps/rabbitmq_management/src/rabbit_mgmt_wm_queues.erl
@@ -76,9 +76,11 @@ basic(ReqData) ->
     %% a cluster wide query with a reasonably long (10s) timeout.
     %% TODO: replace with faster approximate function
     Running = rabbit_nodes:list_running(),
-    Ctx = #{running_nodes => Running},
+    StatsDisabled = rabbit_mgmt_util:disable_stats(ReqData),
+    Ctx = #{running_nodes => Running,
+            management_stats_disabled => StatsDisabled},
     FmtQ = fun (Q) -> rabbit_mgmt_format:queue(Q, Ctx) end,
-    case rabbit_mgmt_util:disable_stats(ReqData) of
+    case StatsDisabled of
         false ->
             list_queues(ReqData, Running, FmtQ, FmtQ);
         true ->
diff --git a/deps/rabbitmq_management/src/rabbit_mgmt_wm_user_queues.erl b/deps/rabbitmq_management/src/rabbit_mgmt_wm_user_queues.erl
@@ -70,7 +70,9 @@ basic(ReqData) ->
     %% a cluster wide query with a reasonably long (10s) timeout.
     %% TODO: replace with faster approximate function
     Running = rabbit_nodes:list_running(),
-    Ctx = #{running_nodes => Running},
+    StatsDisabled = rabbit_mgmt_util:disable_stats(ReqData),
+    Ctx = #{running_nodes => Running,
+            management_stats_disabled => StatsDisabled},
     FmtQ = fun (Q) -> rabbit_mgmt_format:queue(Q, Ctx) end,
     User = rabbit_mgmt_util:id(user, ReqData),
     list_queues(ReqData, Running, FmtQ, FmtQ, User).
diff --git a/deps/rabbitmq_management_agent/src/rabbit_mgmt_format.erl b/deps/rabbitmq_management_agent/src/rabbit_mgmt_format.erl
@@ -15,7 +15,7 @@
 -export([to_amqp_table/1, listener/1, web_context/1, properties/1, basic_properties/1]).
 -export([record/2, to_basic_properties/1]).
 -export([addr/1, port/1]).
--export([format_nulls/1]).
+-export([prepare_for_encoding/1]).
 -export([print/2, print/1]).
 
 -export([format_queue_stats/1, format_queue_basic_stats/1,
@@ -555,23 +555,39 @@ strip_pids([Any | T], Acc) ->
 strip_pids([], Acc) ->
     Acc.
 
-%% Format for JSON replies. Transforms '' into null
-format_nulls(Items) when is_list(Items) ->
-    [format_null_item(Pair) || Pair <- Items];
-format_nulls(Item) ->
+%% Format for JSON replies. Transforms '' into null and deduplicates keys.
+prepare_for_encoding(Items) when is_list(Items) ->
+    dedup_keys([format_null_item(Pair) || Pair <- Items]);
+prepare_for_encoding(Item) ->
     format_null_item(Item).
 
 format_null_item({Key, ''}) ->
     {Key, null};
 format_null_item({Key, Value}) when is_list(Value) ->
-    {Key, format_nulls(Value)};
+    {Key, prepare_for_encoding(Value)};
 format_null_item({Key, Value}) ->
     {Key, Value};
 format_null_item([{_K, _V} | _T] = L) ->
-    format_nulls(L);
+    prepare_for_encoding(L);
 format_null_item(Value) ->
     Value.
 
+%% Keep only the first occurrence of each key in a proplist.
+dedup_keys(Items) ->
+    dedup_keys(Items, sets:new([{version, 2}]), []).
+
+dedup_keys([], _Seen, Acc) ->
+    lists:reverse(Acc);
+dedup_keys([{Key, _} = Item | Rest], Seen, Acc) ->
+    case sets:is_element(Key, Seen) of
+        true ->
+            dedup_keys(Rest, Seen, Acc);
+        false ->
+            dedup_keys(Rest, sets:add_element(Key, Seen), [Item | Acc])
+    end;
+dedup_keys([Item | Rest], Seen, Acc) ->
+    dedup_keys(Rest, Seen, [Item | Acc]).
+
 -spec clean_consumer_details(proplists:proplist()) -> proplists:proplist().
 clean_consumer_details(Obj) ->
      case pget(consumer_details, Obj) of
