Allow MQTT QoS 0 subscribers to reconnect

The solution in #10203 has the following issues:
1. Bindings can be left ofter in Mnesia table rabbit_durable_queue.
One solution to 1. would be to first delete the old queue via
`rabbit_amqqueue:internal_delete(Q, User, missing_owner)`
and subsequently declare the new queue via
`rabbit_amqqueue:internal_declare(Q, false)`
However, even then, it suffers from:
2. Race conditions between `rabbit_amqqueue:on_node_down/1`
and `rabbit_mqtt_qos0_queue:declare/2`:
`rabbit_amqqueue:on_node_down/1` could first read the queue records that
need to be deleted, thereafter `rabbit_mqtt_qos0_queue:declare/2` could
re-create the queue owned by the new connection PID, and `rabbit_amqqueue:on_node_down/1`
could subsequently delete the re-created queue.

Unfortunately, `rabbit_amqqueue:on_node_down/1` does not delete
transient queues in one isolated transaction. Instead it first reads
queues and subsequenlty deletes queues in batches making it prone to
race conditions.

Ideally, this commit deletes all rabbit_mqtt_qos0_queue queues of the
node that has crashed including their bindings.
However, doing so in one transaction is risky as there may be millions
of such queues and the current code path applies the same logic on all
live nodes resulting in conflicting transactions and therefore a long
database operation.

Hence, this commit uses the simplest approach which should still be
safe:
Do not remove rabbit_mqtt_qos0_queue queues if a node crashes.
Other live nodes will continue to route to these dead queues.
That should be okay, given that the rabbit_mqtt_qos0_queue clients auto
confirm.
Continuing routing however has the effect of counting as routing result
for AMQP 0.9.1 `mandatory` property.
If an MQTT client re-connects to a live node with the same client ID,
the new node will delete and then re-create the queue.
Once the crashed node comes back online, it will clean up its leftover
queues and bindings.

(cherry picked from commit 78b4fcc)

# Conflicts:
#	deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl

Author: ansd

deps/rabbit/src/rabbit_amqqueue.erl

@@ -1972,7 +1972,8 @@ on_node_down(Node) ->
         {QueueNames, Deletions} ->
             case length(QueueNames) of
                 0 -> ok;
-                _ -> rabbit_log:info("~tp transient queues from an old incarnation of node ~tp deleted in ~fs", [length(QueueNames), Node, Time/1000000])
+                N -> rabbit_log:info("~b transient queues from an old incarnation of node ~tp deleted in ~fs",
+                                     [N, Node, Time / 1_000_000])
             end,
             notify_queue_binding_deletions(Deletions),
             rabbit_core_metrics:queues_deleted(QueueNames),
@@ -1987,6 +1988,7 @@ filter_transient_queues_to_delete(Node) ->
                 andalso (not amqqueue:is_classic(Q) orelse not amqqueue:is_durable(Q))
                 andalso (not rabbit_amqqueue:is_replicated(Q)
                          orelse rabbit_amqqueue:is_dead_exclusive(Q))
+                andalso amqqueue:get_type(Q) =/= rabbit_mqtt_qos0_queue
     end.
 
 notify_queue_binding_deletions(QueueDeletions) when is_list(QueueDeletions) ->

deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl

@@ -67,7 +67,8 @@ is_stateful() ->
     false.
 
 -spec declare(amqqueue:amqqueue(), node()) ->
-    {'new' | 'existing' | 'owner_died', amqqueue:amqqueue()}.
+    {'new' | 'existing' | 'owner_died', amqqueue:amqqueue()} |
+    {'absent', amqqueue:amqqueue(), rabbit_amqqueue:absent_reason()}.
 declare(Q0, _Node) ->
     %% The queue gets persisted such that routing to this
     %% queue (via the topic exchange) works as usual.
@@ -84,6 +85,7 @@ declare(Q0, _Node) ->
                                  {arguments, amqqueue:get_arguments(Q0)},
                                  {user_who_performed_action, ActingUser}]),
             {new, Q};
+<<<<<<< HEAD
         {absent, OldQ, nodedown} ->
             %% This case body can be deleted once Mnesia is unsupported.
             OldPid = amqqueue:get_pid(OldQ),
@@ -98,6 +100,8 @@ declare(Q0, _Node) ->
                 Other ->
                     Other
             end;
+=======
+>>>>>>> 78b4fcc899 (Allow MQTT QoS 0 subscribers to reconnect)
         Other ->
             Other
     end.

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -1026,38 +1026,45 @@ rabbit_mqtt_qos0_queue(Config) ->
     ok = emqtt:disconnect(Pub).
 
 rabbit_mqtt_qos0_queue_kill_node(Config) ->
-    Topic = atom_to_binary(?FUNCTION_NAME),
+    Topic1 = <<"t/1">>,
+    Topic2 = <<"t/2">>,
     Pub = connect(<<"publisher">>, Config, 2, []),
 
     SubscriberId = <<"subscriber">>,
     Sub0 = connect(SubscriberId, Config, 0, []),
-    {ok, _, [0]} = emqtt:subscribe(Sub0, Topic, qos0),
-    ok = emqtt:publish(Pub, Topic, <<"m0">>, qos0),
-    ok = expect_publishes(Sub0, Topic, [<<"m0">>]),
+    {ok, _, [0]} = emqtt:subscribe(Sub0, Topic1, qos0),
+    ok = emqtt:publish(Pub, Topic1, <<"m0">>, qos0),
+    ok = expect_publishes(Sub0, Topic1, [<<"m0">>]),
 
     process_flag(trap_exit, true),
     ok = rabbit_ct_broker_helpers:kill_node(Config, 0),
-    %% Wait a bit to ensure that Mnesia deletes the queue record on nodes 1 and 2 from Mnesia
-    %% table rabbit_queue (but the queue record is still present in rabbit_durable_queue).
+    ok = await_exit(Sub0),
+    %% Wait to run rabbit_amqqueue:on_node_down/1 on both live nodes.
     timer:sleep(500),
+    %% Re-connect to a live node with same MQTT client ID.
     Sub1 = connect(SubscriberId, Config, 1, []),
-    {ok, _, [0]} = emqtt:subscribe(Sub1, Topic, qos0),
-    ok = emqtt:publish(Pub, Topic, <<"m1">>, qos0),
-    ok = expect_publishes(Sub1, Topic, [<<"m1">>]),
+    {ok, _, [0]} = emqtt:subscribe(Sub1, Topic2, qos0),
+    ok = emqtt:publish(Pub, Topic2, <<"m1">>, qos0),
+    ok = expect_publishes(Sub1, Topic2, [<<"m1">>]),
+    %% Since we started a new clean session, previous subscription should have been deleted.
+    ok = emqtt:publish(Pub, Topic1, <<"m2">>, qos0),
+    receive {publish, _} = Publish -> ct:fail({unexpected, Publish})
+    after 300 -> ok
+    end,
 
-    %% Start node 0 to have a majority for Khepri.
     ok = rabbit_ct_broker_helpers:start_node(Config, 0),
     ok = rabbit_ct_broker_helpers:kill_node(Config, 1),
-    %% This time, do not wait. Mnesia will contain the queue record in rabbit_durable_queue,
-    %% but this time Mnesia may or may not contain the queue record in rabbit_queue.
+    %% This time, do not wait.
+    %% rabbit_amqqueue:on_node_down/1 may or may not have run.
     Sub2 = connect(SubscriberId, Config, 2, []),
-    {ok, _, [0]} = emqtt:subscribe(Sub2, Topic, qos0),
-    ok = emqtt:publish(Pub, Topic, <<"m2">>, qos0),
-    ok = expect_publishes(Sub2, Topic, [<<"m2">>]),
+    {ok, _, [0]} = emqtt:subscribe(Sub2, Topic2, qos0),
+    ok = emqtt:publish(Pub, Topic2, <<"m3">>, qos0),
+    ok = expect_publishes(Sub2, Topic2, [<<"m3">>]),
 
     ok = emqtt:disconnect(Sub2),
     ok = emqtt:disconnect(Pub),
-    ok = rabbit_ct_broker_helpers:start_node(Config, 1).
+    ok = rabbit_ct_broker_helpers:start_node(Config, 1),
+    ?assertEqual([], rpc(Config, rabbit_db_binding, get_all, [])).
 
 %% Test that MQTT connection can be listed and closed via the rabbitmq_management plugin.
 management_plugin_connection(Config) ->