Merge branch 'main' into ra_tick_only_one_pid

SimonUnge · SimonUnge · commit 1f84a14681bd · 2023-03-20T15:07:08.000-07:00
diff --git a/deps/rabbit/BUILD.bazel b/deps/rabbit/BUILD.bazel
@@ -78,6 +78,8 @@ _APP_ENV = """[
 	                         ]},
 	    {halt_on_upgrade_failure, true},
 	    {ssl_apps, [asn1, crypto, public_key, ssl]},
+	    %% classic queue storage implementation version
+	    {classic_queue_default_version, 2},
 	    %% see rabbitmq-server#114
 	    {mirroring_flow_control, true},
 	    {mirroring_sync_batch_size, 4096},
diff --git a/deps/rabbit/Makefile b/deps/rabbit/Makefile
@@ -65,6 +65,8 @@ define PROJECT_ENV
 	                         ]},
 	    {halt_on_upgrade_failure, true},
 	    {ssl_apps, [asn1, crypto, public_key, ssl]},
+	    %% classic queue storage implementation version
+        {classic_queue_default_version, 2},
 	    %% see rabbitmq-server#114
 	    {mirroring_flow_control, true},
 	    {mirroring_sync_batch_size, 4096},
diff --git a/deps/rabbit/priv/schema/rabbit.schema b/deps/rabbit/priv/schema/rabbit.schema
@@ -2408,7 +2408,7 @@ end}.
 
 {translation, "rabbit.classic_queue_default_version",
     fun(Conf) ->
-        case cuttlefish:conf_get("classic_queue.default_version", Conf, 1) of
+        case cuttlefish:conf_get("classic_queue.default_version", Conf, 2) of
             1 -> 1;
             2 -> 2;
             _ -> cuttlefish:unset()
diff --git a/deps/rabbit/src/rabbit_amqqueue.erl b/deps/rabbit/src/rabbit_amqqueue.erl
@@ -133,14 +133,6 @@ recover(VHost) ->
 filter_pid_per_type(QPids) ->
     lists:partition(fun(QPid) -> ?IS_CLASSIC(QPid) end, QPids).
 
-filter_resource_per_type(Resources) ->
-    Queues = [begin
-                  {ok, Q} = lookup(Resource),
-                  QPid = amqqueue:get_pid(Q),
-                  {Resource, QPid}
-              end || Resource <- Resources],
-    lists:partition(fun({_Resource, QPid}) -> ?IS_CLASSIC(QPid) end, Queues).
-
 -spec stop(rabbit_types:vhost()) -> 'ok'.
 stop(VHost) ->
     %% Classic queues
@@ -1506,11 +1498,17 @@ delete_immediately(QPids) ->
     end.
 
 delete_immediately_by_resource(Resources) ->
-    {Classic, Quorum} = filter_resource_per_type(Resources),
-    [gen_server2:cast(QPid, delete_immediately) || {_, QPid} <- Classic],
-    [rabbit_quorum_queue:delete_immediately(Resource, QPid)
-     || {Resource, QPid} <- Quorum],
-    ok.
+    lists:foreach(
+      fun(Resource) ->
+              {ok, Q} = lookup(Resource),
+              QPid = amqqueue:get_pid(Q),
+              case ?IS_CLASSIC(QPid) of
+                  true ->
+                      gen_server2:cast(QPid, delete_immediately);
+                  _ ->
+                      rabbit_quorum_queue:delete_immediately(Q)
+              end
+      end, Resources).
 
 -spec delete
         (amqqueue:amqqueue(), 'false', 'false', rabbit_types:username()) ->
@@ -1676,12 +1674,13 @@ notify_sent_queue_down(QPid) ->
 resume(QPid, ChPid) -> delegate:invoke_no_result(QPid, {gen_server2, cast,
                                                         [{resume, ChPid}]}).
 
--spec internal_delete(name(), rabbit_types:username()) -> 'ok'.
+-spec internal_delete(amqqueue:amqqueue(), rabbit_types:username()) -> 'ok'.
 
-internal_delete(QueueName, ActingUser) ->
-    internal_delete(QueueName, ActingUser, normal).
+internal_delete(Queue, ActingUser) ->
+    internal_delete(Queue, ActingUser, normal).
 
-internal_delete(QueueName, ActingUser, Reason) ->
+internal_delete(Queue, ActingUser, Reason) ->
+    QueueName = amqqueue:get_name(Queue),
     case rabbit_db_queue:delete(QueueName, Reason) of
         ok ->
             ok;
@@ -1691,6 +1690,7 @@ internal_delete(QueueName, ActingUser, Reason) ->
             rabbit_core_metrics:queue_deleted(QueueName),
             ok = rabbit_event:notify(queue_deleted,
                                      [{name, QueueName},
+                                      {type, amqqueue:get_type(Queue)},
                                       {user_who_performed_action, ActingUser}])
     end.
 
@@ -1874,7 +1874,7 @@ on_node_down(Node) ->
             end,
             notify_queue_binding_deletions(Deletions),
             rabbit_core_metrics:queues_deleted(QueueNames),
-            notify_queues_deleted(QueueNames),
+            notify_transient_queues_deleted(QueueNames),
             ok
     end.
 
@@ -1897,11 +1897,12 @@ notify_queue_binding_deletions(QueueDeletions) ->
     Deletions = rabbit_binding:process_deletions(QueueDeletions),
     rabbit_binding:notify_deletions(Deletions, ?INTERNAL_USER).
 
-notify_queues_deleted(QueueDeletions) ->
+notify_transient_queues_deleted(QueueDeletions) ->
     lists:foreach(
       fun(Queue) ->
               ok = rabbit_event:notify(queue_deleted,
                                        [{name, Queue},
+                                        {kind, rabbit_classic_queue},
                                         {user, ?INTERNAL_USER}])
       end,
       QueueDeletions).
diff --git a/deps/rabbit/src/rabbit_amqqueue_process.erl b/deps/rabbit/src/rabbit_amqqueue_process.erl
@@ -314,7 +314,6 @@ terminate_delete(EmitStats, Reason0,
                  State = #q{q = Q,
                             backing_queue = BQ,
                             status = Status}) ->
-    QName = amqqueue:get_name(Q),
     ActingUser = terminated_by(Status),
     fun (BQS) ->
         Reason = case Reason0 of
@@ -330,7 +329,7 @@ terminate_delete(EmitStats, Reason0,
         %% logged.
         try
             %% don't care if the internal delete doesn't return 'ok'.
-            rabbit_amqqueue:internal_delete(QName, ActingUser, Reason0)
+            rabbit_amqqueue:internal_delete(Q, ActingUser, Reason0)
         catch
             {error, ReasonE} -> error(ReasonE)
         end,
@@ -471,12 +470,10 @@ init_queue_mode(Mode, State = #q {backing_queue = BQ,
 
 init_queue_version(Version0, State = #q {backing_queue = BQ,
                                          backing_queue_state = BQS}) ->
-    %% When the version is undefined we use the default version 1.
-    %% We want to BQ:set_queue_version in all cases because a v2
-    %% policy might have been deleted, for example, and we want
-    %% the queue to go back to v1.
+    %% When the version is undefined we use the default version 2 starting with
+    %% RabbitMQ 3.12.0.
     Version = case Version0 of
-        undefined -> rabbit_misc:get_env(rabbit, classic_queue_default_version, 1);
+        undefined -> rabbit_misc:get_env(rabbit, classic_queue_default_version, 2);
         _ -> Version0
     end,
     BQS1 = BQ:set_queue_version(Version, BQS),
diff --git a/deps/rabbit/src/rabbit_classic_queue.erl b/deps/rabbit/src/rabbit_classic_queue.erl
@@ -466,10 +466,9 @@ delete_crashed(Q, ActingUser) ->
                   [Q, ActingUser]).
 
 delete_crashed_internal(Q, ActingUser) ->
-    QName = amqqueue:get_name(Q),
     {ok, BQ} = application:get_env(rabbit, backing_queue_module),
     BQ:delete_crashed(Q),
-    ok = rabbit_amqqueue:internal_delete(QName, ActingUser).
+    ok = rabbit_amqqueue:internal_delete(Q, ActingUser).
 
 recover_durable_queues(QueuesAndRecoveryTerms) ->
     {Results, Failures} =
diff --git a/deps/rabbit/src/rabbit_fifo_dlx.erl b/deps/rabbit/src/rabbit_fifo_dlx.erl
@@ -114,11 +114,20 @@ apply(_, {dlx, #checkout{consumer = Pid,
 apply(_, {dlx, #checkout{consumer = ConsumerPid,
                          prefetch = Prefetch}},
       at_least_once,
-      #?MODULE{consumer = #dlx_consumer{checked_out = CheckedOutOldConsumer},
+      #?MODULE{consumer = #dlx_consumer{pid = OldConsumerPid,
+                                        checked_out = CheckedOutOldConsumer},
                discards = Discards0,
                msg_bytes = Bytes,
                msg_bytes_checkout = BytesCheckout} = State0) ->
     %% Since we allow only a single consumer, the new consumer replaces the old consumer.
+    case ConsumerPid of
+        OldConsumerPid ->
+            ok;
+        _ ->
+            rabbit_log:debug("Terminating ~p since ~p becomes active rabbit_fifo_dlx_worker",
+                             [OldConsumerPid, ConsumerPid]),
+            ensure_worker_terminated(State0)
+    end,
     %% All checked out messages to the old consumer need to be returned to the discards queue
     %% such that these messages will be re-delivered to the new consumer.
     %% When inserting back into the discards queue, we respect the original order in which messages
diff --git a/deps/rabbit/src/rabbit_fifo_dlx_client.erl b/deps/rabbit/src/rabbit_fifo_dlx_client.erl
@@ -32,7 +32,7 @@ settle(MsgIds, #state{leader = Leader} = State)
     {ok, State}.
 
 -spec checkout(rabbit_amqqueue:name(), ra:server_id(), non_neg_integer()) ->
-    {ok, state()} | {error, ra_command_failed}.
+    {ok, state()} | {error, non_local_leader | ra_command_failed}.
 checkout(QResource, Leader, NumUnsettled) ->
     Cmd = rabbit_fifo_dlx:make_checkout(self(), NumUnsettled),
     State = #state{queue_resource = QResource,
@@ -46,10 +46,10 @@ process_command(Cmd, #state{leader = Leader} = State, Tries) ->
     case ra:process_command(Leader, Cmd, 60_000) of
         {ok, ok, Leader} ->
             {ok, State#state{leader = Leader}};
-        {ok, ok, L} ->
+        {ok, ok, NonLocalLeader} ->
             rabbit_log:warning("Failed to process command ~tp on quorum queue leader ~tp because actual leader is ~tp.",
-                               [Cmd, Leader, L]),
-            {error, ra_command_failed};
+                               [Cmd, Leader, NonLocalLeader]),
+            {error, non_local_leader};
         Err ->
             rabbit_log:warning("Failed to process command ~tp on quorum queue leader ~tp: ~tp~n"
                                "Trying ~b more time(s)...",
diff --git a/deps/rabbit/src/rabbit_fifo_dlx_sup.erl b/deps/rabbit/src/rabbit_fifo_dlx_sup.erl
@@ -24,11 +24,12 @@ start_link() ->
 
 init([]) ->
     SupFlags = #{strategy => simple_one_for_one,
-                 intensity => 1,
-                 period => 5},
+                 intensity => 100,
+                 period => 1},
     Worker = rabbit_fifo_dlx_worker,
     ChildSpec = #{id => Worker,
                   start => {Worker, start_link, []},
                   type => worker,
+                  restart => transient,
                   modules => [Worker]},
     {ok, {SupFlags, [ChildSpec]}}.
diff --git a/deps/rabbit/src/rabbit_fifo_dlx_worker.erl b/deps/rabbit/src/rabbit_fifo_dlx_worker.erl
@@ -106,23 +106,27 @@ init(QRef) ->
     {ok, undefined, {continue, QRef}}.
 
 -spec handle_continue(rabbit_amqqueue:name(), undefined) ->
-    {noreply, state()}.
+    {noreply, state()} | {stop, term(), undefined}.
 handle_continue(QRef, undefined) ->
     {ok, Prefetch} = application:get_env(rabbit,
                                          dead_letter_worker_consumer_prefetch),
     {ok, SettleTimeout} = application:get_env(rabbit,
                                               dead_letter_worker_publisher_confirm_timeout),
     {ok, Q} = rabbit_amqqueue:lookup(QRef),
     {ClusterName, _MaybeOldLeaderNode} = amqqueue:get_pid(Q),
-    {ok, ConsumerState} = rabbit_fifo_dlx_client:checkout(QRef,
-                                                          {ClusterName, node()},
-                                                          Prefetch),
-    {noreply, lookup_topology(#state{queue_ref = QRef,
-                                     queue_type_state = rabbit_queue_type:init(),
-                                     settle_timeout = SettleTimeout,
-                                     dlx_client_state = ConsumerState,
-                                     monitor_ref = erlang:monitor(process, ClusterName)
-                                    })}.
+    case rabbit_fifo_dlx_client:checkout(QRef, {ClusterName, node()}, Prefetch) of
+        {ok, ConsumerState} ->
+            {noreply, lookup_topology(#state{queue_ref = QRef,
+                                             queue_type_state = rabbit_queue_type:init(),
+                                             settle_timeout = SettleTimeout,
+                                             dlx_client_state = ConsumerState,
+                                             monitor_ref = erlang:monitor(process, ClusterName)
+                                            })};
+        {error, non_local_leader = Reason} ->
+            {stop, {shutdown, Reason}, undefined};
+        Error ->
+            {stop, Error, undefined}
+    end.
 
 terminate(_Reason, State) ->
     cancel_timer(State).
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -20,7 +20,7 @@
          restart_server/1,
          stop_server/1,
          delete/4,
-         delete_immediately/2]).
+         delete_immediately/1]).
 -export([state_info/1, info/2, stat/1, infos/1]).
 -export([settle/5, dequeue/5, consume/3, cancel/5]).
 -export([credit/5]).
@@ -223,20 +223,20 @@ start_cluster(Q) ->
                                           ActingUser}]),
                     {new, NewQ};
                 {error, Error} ->
-                    declare_queue_error(Error, QName, Leader, ActingUser)
+                    declare_queue_error(Error, NewQ, Leader, ActingUser)
             catch
                 error:Error ->
-                    declare_queue_error(Error, QName, Leader, ActingUser)
+                    declare_queue_error(Error, NewQ, Leader, ActingUser)
             end;
         {existing, _} = Ex ->
             Ex
     end.
 
-declare_queue_error(Error, QName, Leader, ActingUser) ->
-    _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+declare_queue_error(Error, Queue, Leader, ActingUser) ->
+    _ = rabbit_amqqueue:internal_delete(Queue, ActingUser),
     {protocol_error, internal_error,
      "Cannot declare quorum ~ts on node '~ts' with leader on node '~ts': ~255p",
-     [rabbit_misc:rs(QName), node(), Leader, Error]}.
+     [rabbit_misc:rs(amqqueue:get_name(Queue)), node(), Leader, Error]}.
 
 ra_machine(Q) ->
     {module, rabbit_fifo, ra_machine_config(Q)}.
@@ -660,7 +660,7 @@ delete(Q, _IfUnused, _IfEmpty, ActingUser) when ?amqqueue_is_quorum(Q) ->
                     ok = force_delete_queue(Servers)
             end,
             notify_decorators(QName, shutdown),
-            ok = delete_queue_data(QName, ActingUser),
+            ok = delete_queue_data(Q, ActingUser),
             _ = erpc:call(LeaderNode, rabbit_core_metrics, queue_deleted, [QName],
                           ?RPC_TIMEOUT),
             {ok, ReadyMsgs};
@@ -671,7 +671,7 @@ delete(Q, _IfUnused, _IfEmpty, ActingUser) when ?amqqueue_is_quorum(Q) ->
                 true ->
                     %% If all ra nodes were already down, the delete
                     %% has succeed
-                    delete_queue_data(QName, ActingUser),
+                    delete_queue_data(Q, ActingUser),
                     {ok, ReadyMsgs};
                 false ->
                     %% attempt forced deletion of all servers
@@ -682,7 +682,7 @@ delete(Q, _IfUnused, _IfEmpty, ActingUser) when ?amqqueue_is_quorum(Q) ->
                       [rabbit_misc:rs(QName), Errs]),
                     ok = force_delete_queue(Servers),
                     notify_decorators(QName, shutdown),
-                    delete_queue_data(QName, ActingUser),
+                    delete_queue_data(Q, ActingUser),
                     {ok, ReadyMsgs}
             end
     end.
@@ -701,15 +701,15 @@ force_delete_queue(Servers) ->
      end || S <- Servers],
     ok.
 
-delete_queue_data(QName, ActingUser) ->
-    _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+delete_queue_data(Queue, ActingUser) ->
+    _ = rabbit_amqqueue:internal_delete(Queue, ActingUser),
     ok.
 
 
-delete_immediately(Resource, {_Name, _} = QPid) ->
-    _ = rabbit_amqqueue:internal_delete(Resource, ?INTERNAL_USER),
-    {ok, _} = ra:delete_cluster([QPid]),
-    rabbit_core_metrics:queue_deleted(Resource),
+delete_immediately(Queue) ->
+    _ = rabbit_amqqueue:internal_delete(Queue, ?INTERNAL_USER),
+    {ok, _} = ra:delete_cluster([amqqueue:get_pid(Queue)]),
+    rabbit_core_metrics:queue_deleted(amqqueue:get_name(Queue)),
     ok.
 
 settle(_QName, complete, CTag, MsgIds, QState) ->
diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl
@@ -177,8 +177,7 @@ delete_stream(Q, ActingUser)
     #{name := StreamId} = amqqueue:get_type_state(Q),
     case process_command({delete_stream, StreamId, #{}}) of
         {ok, ok, _} ->
-            QName = amqqueue:get_name(Q),
-              _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+            _ = rabbit_amqqueue:internal_delete(Q, ActingUser),
             {ok, {ok, 0}};
         Err ->
             Err
diff --git a/deps/rabbit/src/rabbit_stream_queue.erl b/deps/rabbit/src/rabbit_stream_queue.erl
@@ -160,8 +160,7 @@ create_stream(Q0) ->
                                           ActingUser}]),
                     {new, Q};
                 Error ->
-
-                    _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+                    _ = rabbit_amqqueue:internal_delete(Q, ActingUser),
                     {protocol_error, internal_error, "Cannot declare a queue '~ts' on node '~ts': ~255p",
                      [rabbit_misc:rs(QName), node(), Error]}
             end;
diff --git a/deps/rabbit/src/rabbit_variable_queue.erl b/deps/rabbit/src/rabbit_variable_queue.erl
@@ -492,8 +492,9 @@ process_recovery_terms(Terms) ->
 
 queue_version(Q) ->
     Resolve = fun(_, ArgVal) -> ArgVal end,
+    %% If queue-version is undefined, we assume v2 starting with RabbitMQ 3.12.0.
     case rabbit_queue_type_util:args_policy_lookup(<<"queue-version">>, Resolve, Q) of
-        undefined -> rabbit_misc:get_env(rabbit, classic_queue_default_version, 1);
+        undefined -> rabbit_misc:get_env(rabbit, classic_queue_default_version, 2);
         Vsn when is_integer(Vsn) -> Vsn;
         Vsn -> binary_to_integer(Vsn)
     end.
diff --git a/deps/rabbit/src/rabbit_vhost.erl b/deps/rabbit/src/rabbit_vhost.erl
@@ -413,8 +413,7 @@ assert_benign({error, not_found}, _) -> ok;
 assert_benign({error, {absent, Q, _}}, ActingUser) ->
     %% Removing the database entries here is safe. If/when the down node
     %% restarts, it will clear out the on-disk storage of the queue.
-    QName = amqqueue:get_name(Q),
-    rabbit_amqqueue:internal_delete(QName, ActingUser).
+    rabbit_amqqueue:internal_delete(Q, ActingUser).
 
 -spec exists(vhost:name()) -> boolean().
 
diff --git a/deps/rabbit/test/backing_queue_SUITE.erl b/deps/rabbit/test/backing_queue_SUITE.erl
@@ -911,7 +911,7 @@ bq_queue_recover1(Config) ->
                   rabbit_variable_queue:fetch(true, VQ1),
               CountMinusOne = rabbit_variable_queue:len(VQ2),
               _VQ3 = rabbit_variable_queue:delete_and_terminate(shutdown, VQ2),
-              ok = rabbit_amqqueue:internal_delete(QName, <<"acting-user">>)
+              ok = rabbit_amqqueue:internal_delete(Q1, <<"acting-user">>)
       end),
     passed.
 
diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt_qos0_queue.erl
diff --git a/release-notes/3.11.11.md b/release-notes/3.11.11.md
diff --git a/release-notes/3.12.0.md b/release-notes/3.12.0.md
