Keep storing MQTT client IDs as lists in Ra

Up to 3.11.x an MQTT client ID is tracked in Ra
as a list of bytes as returned by binary_to_list/1 in
https://github.com/rabbitmq/rabbitmq-server/blob/48467d6e1283b8d81e52cfd49c06ea4eaa31617d/deps/rabbitmq_mqtt/src/rabbit_mqtt_frame.erl#L137

This has two downsides:
1. Lists consume more memory than binaries (when tracking many clients).
2. It violates the MQTT spec which states
   "The ClientId MUST be a UTF-8 encoded string as defined in Section 1.5.3 [MQTT-3.1.3-4]." [v4 3.1.3.1]

Therefore, the original idea was to always store MQTT client IDs as
binaries starting with Native MQTT in 3.12.
However, this leads to client ID tracking misbehaving in mixed version
clusters since new nodes would register client IDs as binaries and old
nodes would register client IDs as lists. This means that a client
registering on a new node with the same client ID as a connection to the
old node did not terminate the connection on the old node.

Therefore, for backwards compatibility, we leave the client ID as a list of bytes
in the Ra machine state because the feature flag delete_ra_cluster_mqtt_node
introduced in v3.12 will delete the Ra cluster anyway and
the new client ID tracking via pg local will store client IDs as
binaries.

An interesting side note learned here is that the compiled file
rabbit_mqtt_collector must not be changed. This commit only modifies
function specs. However as soon as the compiled code is changed, this
module becomes a new version. The new version causes the anonymous ra query
function to fail in mixed clusters: When the old node does a
ra:leader_query where the leader is on the new node, the query function
fails on the new node with `badfun` because the new node does not have
the same module version. For more context, read:
https://web.archive.org/web/20181017104411/http://www.javalimit.com/2010/05/passing-funs-to-other-erlang-nodes.html

Author: ansd

deps/rabbitmq_mqtt/include/mqtt_machine.hrl

@@ -5,12 +5,21 @@
 %% Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 %%
 
+%% A client ID that is tracked in Ra is a list of bytes
+%% as returned by binary_to_list/1 in
+%% https://github.com/rabbitmq/rabbitmq-server/blob/48467d6e1283b8d81e52cfd49c06ea4eaa31617d/deps/rabbitmq_mqtt/src/rabbit_mqtt_frame.erl#L137
+%% prior to 3.12.0.
+%% This has two downsides:
+%% 1. Lists consume more memory than binaries (when tracking many clients).
+%% 2. This violates the MQTT spec which states
+%%    "The ClientId MUST be a UTF-8 encoded string as defined in Section 1.5.3 [MQTT-3.1.3-4]." [v4 3.1.3.1]
+%% However, for backwards compatibility, we leave the client ID as a list of bytes in the Ra machine state because
+%% feature flag delete_ra_cluster_mqtt_node introduced in 3.12.0 will delete the Ra cluster anyway.
+-type client_id() :: [byte()].
+
 -record(machine_state, {
-          %% client ID to connection PID
-          client_ids = #{},
-          %% connection PID to list of client IDs
-          pids = #{},
+          client_ids = #{} :: #{client_id() => Connection :: pid()},
+          pids = #{} :: #{Connection :: pid() => [client_id(), ...]},
           %% add acouple of fields for future extensibility
           reserved_1,
           reserved_2}).
-

deps/rabbitmq_mqtt/src/mqtt_machine.erl

@@ -22,7 +22,6 @@
 -type config() :: map().
 
 -type reply() :: {ok, term()} | {error, term()}.
--type client_id() :: term().
 
 -type command() :: {register, client_id(), pid()} |
                    {unregister, client_id(), pid()} |

deps/rabbitmq_mqtt/src/rabbit_mqtt_collector.erl

@@ -13,7 +13,7 @@
          list/0, list_pids/0, leave/1]).
 
 %%----------------------------------------------------------------------------
--spec register(term(), pid()) -> {ok, reference()} | {error, term()}.
+-spec register(client_id(), pid()) -> {ok, reference()} | {error, term()}.
 register(ClientId, Pid) ->
     {ClusterName, _} = NodeId = mqtt_node:server_id(),
     case ra_leaderboard:lookup_leader(ClusterName) of
@@ -28,15 +28,15 @@ register(ClientId, Pid) ->
             register(Leader, ClientId, Pid)
     end.
 
--spec register(ra:server_id(), term(), pid()) ->
+-spec register(ra:server_id(), client_id(), pid()) ->
     {ok, reference()} | {error, term()}.
 register(ServerId, ClientId, Pid) ->
     Corr = make_ref(),
     send_ra_command(ServerId, {register, ClientId, Pid}, Corr),
     erlang:send_after(5000, self(), {ra_event, undefined, register_timeout}),
     {ok, Corr}.
 
--spec unregister(binary(), pid()) -> ok.
+-spec unregister(client_id(), pid()) -> ok.
 unregister(ClientId, Pid) ->
     {ClusterName, _} = mqtt_node:server_id(),
     case ra_leaderboard:lookup_leader(ClusterName) of

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -426,7 +426,7 @@ register_client_id(VHost, ClientId)
 
     case rabbit_mqtt_ff:track_client_id_in_ra() of
         true ->
-            case rabbit_mqtt_collector:register(ClientId, self()) of
+            case collector_register(ClientId) of
                 {ok, Corr} ->
                     %% Ra node takes care of removing duplicate client ID connections.
                     {ok, {pending, Corr}};
@@ -1228,7 +1228,7 @@ maybe_send_will(_, _, _) ->
 unregister_client(#state{cfg = #cfg{client_id = ClientId}}) ->
     case rabbit_mqtt_ff:track_client_id_in_ra() of
         true ->
-            rabbit_mqtt_collector:unregister(ClientId, self());
+            rabbit_mqtt_collector:unregister(binary_to_list(ClientId), self());
         false ->
             ok
     end.
@@ -1283,9 +1283,10 @@ handle_ra_event({applied, [{Corr, ok}]},
     State#state{ra_register_state = registered};
 handle_ra_event({not_leader, Leader, Corr},
                 State = #state{ra_register_state = {pending, Corr},
-                               cfg = #cfg{client_id = ClientId}}) ->
+                               cfg = #cfg{client_id = ClientIdBin}}) ->
     case rabbit_mqtt_ff:track_client_id_in_ra() of
         true ->
+            ClientId = binary_to_list(ClientIdBin),
             %% retry command against actual leader
             {ok, NewCorr} = rabbit_mqtt_collector:register(Leader, ClientId, self()),
             State#state{ra_register_state = {pending, NewCorr}};
@@ -1297,7 +1298,7 @@ handle_ra_event(register_timeout,
                                cfg = #cfg{client_id = ClientId}}) ->
     case rabbit_mqtt_ff:track_client_id_in_ra() of
         true ->
-            {ok, NewCorr} = rabbit_mqtt_collector:register(ClientId, self()),
+            {ok, NewCorr} = collector_register(ClientId),
             State#state{ra_register_state = {pending, NewCorr}};
         false ->
             State
@@ -1784,6 +1785,10 @@ message_redelivered(true, ProtoVer, QType) ->
 message_redelivered(_, _, _) ->
     ok.
 
+collector_register(ClientIdBin) ->
+    ClientId = binary_to_list(ClientIdBin),
+    rabbit_mqtt_collector:register(ClientId, self()).
+
 -spec format_status(state()) -> map().
 format_status(
   #state{queue_states = QState,

deps/rabbitmq_mqtt/test/cluster_SUITE.erl

@@ -19,10 +19,13 @@
          teardown_steps/0,
          get_node_config/3,
          rabbitmqctl/3,
-         rpc/4,
+         rpc/4, rpc/5,
          stop_node/2
         ]).
 
+-import(rabbit_ct_helpers,
+        [eventually/3]).
+
 -define(OPTS, [{connect_timeout, 1},
                {ack_timeout, 1}]).
 
@@ -127,17 +130,15 @@ connection_id_tracking(Config) ->
     ok = emqtt:disconnect(C3).
 
 connection_id_tracking_on_nodedown(Config) ->
-    Server = get_node_config(Config, 0, nodename),
     C = connect(<<"simpleClient">>, Config, ?OPTS),
     {ok, _, _} = emqtt:subscribe(C, <<"TopicA">>, qos0),
     ok = emqtt:publish(C, <<"TopicA">>, <<"Payload">>),
     ok = expect_publishes(C, <<"TopicA">>, [<<"Payload">>]),
     assert_connection_count(Config, 4, 1),
     process_flag(trap_exit, true),
-    ok = stop_node(Config, Server),
+    ok = stop_node(Config, 0),
     await_exit(C),
-    assert_connection_count(Config, 4, 0),
-    ok.
+    ok = eventually(?_assertEqual([], util:all_connection_pids(1, Config)), 500, 4).
 
 connection_id_tracking_with_decommissioned_node(Config) ->
     case rpc(Config, rabbit_mqtt_ff, track_client_id_in_ra, []) of

deps/rabbitmq_mqtt/test/ff_SUITE.erl

@@ -83,13 +83,13 @@ delete_ra_cluster_mqtt_node(Config) ->
                  rabbit_ct_broker_helpers:enable_feature_flag(Config, FeatureFlag)),
 
     %% Ra processes should be gone
-    rabbit_ct_helpers:eventually(
+    eventually(
       ?_assert(lists:all(fun(Pid) -> Pid =:= undefined end,
                          rabbit_ct_broker_helpers:rpc_all(Config, erlang, whereis, [mqtt_node])))),
     %% new client ID tracking works
     ?assertEqual(1, length(util:all_connection_pids(Config))),
-    ?assert(erlang:is_process_alive(C)),
-    ok = emqtt:disconnect(C).
+    ok = emqtt:disconnect(C),
+    eventually(?_assertEqual(0, length(util:all_connection_pids(Config)))).
 
 rabbit_mqtt_qos0_queue(Config) ->
     FeatureFlag = ?FUNCTION_NAME,

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -87,7 +87,6 @@ subgroups() ->
          ,large_message_amqp_to_mqtt
          ,keepalive
          ,keepalive_turned_off
-         ,duplicate_client_id
          ,block
          ,amqp_to_mqtt_qos0
          ,clean_session_disconnect_client
@@ -112,7 +111,8 @@ subgroups() ->
        maintenance,
        delete_create_queue,
        publish_to_all_queue_types_qos0,
-       publish_to_all_queue_types_qos1
+       publish_to_all_queue_types_qos1,
+       duplicate_client_id
       ]}
     ].
 
@@ -1143,17 +1143,19 @@ keepalive_turned_off(Config) ->
     ok = emqtt:disconnect(C).
 
 duplicate_client_id(Config) ->
+    [Server1, Server2, _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     DuplicateClientId = ?FUNCTION_NAME,
-    C1 = connect(DuplicateClientId, Config),
+    %% Connect to old node in mixed version cluster.
+    C1 = connect(DuplicateClientId, Config, Server2, []),
     eventually(?_assertEqual(1, length(all_connection_pids(Config)))),
-
     process_flag(trap_exit, true),
-    C2 = connect(DuplicateClientId, Config),
+    %% Connect to new node in mixed version cluster.
+    C2 = connect(DuplicateClientId, Config, Server1, []),
     await_exit(C1),
     timer:sleep(200),
     ?assertEqual(1, length(all_connection_pids(Config))),
-
-    ok = emqtt:disconnect(C2).
+    ok = emqtt:disconnect(C2),
+    eventually(?_assertEqual(0, length(all_connection_pids(Config)))).
 
 block(Config) ->
     Topic = ClientId = atom_to_binary(?FUNCTION_NAME),

deps/rabbitmq_mqtt/test/util.erl

@@ -5,6 +5,7 @@
 -include_lib("eunit/include/eunit.hrl").
 
 -export([all_connection_pids/1,
+         all_connection_pids/2,
          publish_qos1_timeout/4,
          sync_publish_result/3,
          get_global_counters/2,
@@ -23,13 +24,18 @@
         ]).
 
 all_connection_pids(Config) ->
-    Nodes = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
-    Result = erpc:multicall(Nodes, rabbit_mqtt, local_connection_pids, [], 5000),
-    lists:foldl(fun({ok, Pids}, Acc) ->
-                        Pids ++ Acc;
-                   (_, Acc) ->
-                        Acc
-                end, [], Result).
+    all_connection_pids(0, Config).
+
+all_connection_pids(Node, Config) ->
+    case rabbit_ct_broker_helpers:rpc(
+           Config, Node, rabbit_feature_flags, is_enabled, [delete_ra_cluster_mqtt_node]) of
+        true ->
+            Nodes = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+            Result = erpc:multicall(Nodes, rabbit_mqtt, local_connection_pids, [], 5000),
+            lists:append([Pids || {ok, Pids} <- Result]);
+        false ->
+            rabbit_ct_broker_helpers:rpc(Config, Node, rabbit_mqtt_collector, list_pids, [])
+    end.
 
 publish_qos1_timeout(Client, Topic, Payload, Timeout) ->
     Mref = erlang:monitor(process, Client),