Per-vhost supervision trees for queues and message stores.

Per-vhost message stores can be restarted, but queues contain
references for old message stores in message store client data,
also queues rely on message store process to report confirms for
messages on disk.
Because after message store restart queues will not get any confirms and
will fail with badarg error trying to access message store with an old client,
queue processes should be restarted together with message stores.
Queue process cannot monitor message store because of backing_queue mechanism,
so they should be controlled by a supervision tree. One tree will contain
queues supervisor and message store proecesses.
Per-vhost supervisor will restart if any of it's children dies.
Per-vhost supervisor restart process will do queue and message store data recovery
the same way as pre-3.7 global message store did, just with VHost as an argument and
in a vhost data directory.

Author: Daniil Fedotov

src/rabbit.erl

@@ -813,10 +813,7 @@ boot_delegate() ->
 
 recover() ->
     rabbit_policy:recover(),
-    Qs = rabbit_amqqueue:recover(),
-    ok = rabbit_binding:recover(rabbit_exchange:recover(),
-                                [QName || #amqqueue{name = QName} <- Qs]),
-    rabbit_amqqueue:start(Qs).
+    rabbit_vhost:recover().
 
 maybe_insert_default_data() ->
     case rabbit_table:needs_default_data() of

src/rabbit_amqqueue_sup_sup.erl

@@ -19,6 +19,7 @@
 -behaviour(supervisor2).
 
 -export([start_link/0, start_queue_process/3]).
+-export([start_for_vhost/1, stop_for_vhost/1, find_for_vhost/2]).
 
 -export([init/1]).
 
@@ -36,14 +37,35 @@
 %%----------------------------------------------------------------------------
 
 start_link() ->
-    supervisor2:start_link({local, ?SERVER}, ?MODULE, []).
+    supervisor2:start_link(?MODULE, []).
 
 start_queue_process(Node, Q, StartMode) ->
-    {ok, _SupPid, QPid} = supervisor2:start_child(
-                            {?SERVER, Node}, [Q, StartMode]),
+    #amqqueue{name = #resource{virtual_host = VHost}} = Q,
+    {ok, Sup} = find_for_vhost(VHost, Node),
+    {ok, _SupPid, QPid} = supervisor2:start_child(Sup, [Q, StartMode]),
     QPid.
 
 init([]) ->
     {ok, {{simple_one_for_one, 10, 10},
           [{rabbit_amqqueue_sup, {rabbit_amqqueue_sup, start_link, []},
             temporary, ?SUPERVISOR_WAIT, supervisor, [rabbit_amqqueue_sup]}]}}.
+
+find_for_vhost(VHost, Node) ->
+    {ok, VHostSup} = rabbit_vhost_sup_sup:vhost_sup(VHost, Node),
+    case supervisor2:find_child(VHostSup, rabbit_amqqueue_sup_sup) of
+        [QSup] -> {ok, QSup};
+        Result -> {error, {queue_supervisor_not_found, Result}}
+    end.
+
+start_for_vhost(VHost) ->
+    {ok, VHostSup} = rabbit_vhost_sup_sup:vhost_sup(VHost),
+    supervisor2:start_child(
+        VHostSup,
+        {rabbit_amqqueue_sup_sup,
+         {rabbit_amqqueue_sup_sup, start_link, []},
+         transient, infinity, supervisor, [rabbit_amqqueue_sup_sup]}).
+
+stop_for_vhost(VHost) ->
+    {ok, VHostSup} = rabbit_vhost_sup_sup:vhost_sup(VHost),
+    ok = supervisor2:terminate_child(VHostSup, rabbit_amqqueue_sup_sup),
+    ok = supervisor2:delete_child(VHostSup, rabbit_amqqueue_sup_sup).

src/rabbit_exchange.erl

@@ -18,7 +18,7 @@
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
 
--export([recover/0, policy_changed/2, callback/4, declare/7,
+-export([recover/1, policy_changed/2, callback/4, declare/7,
          assert_equivalence/6, assert_args_equivalence/2, check_type/1,
          lookup/1, lookup_or_die/1, list/0, list/1, lookup_scratch/2,
          update_scratch/3, update_decorators/1, immutable/1,
@@ -36,7 +36,7 @@
 -type type() :: atom().
 -type fun_name() :: atom().
 
--spec recover() -> [name()].
+-spec recover(rabbit_types:vhost()) -> [name()].
 -spec callback
         (rabbit_types:exchange(), fun_name(),
          fun((boolean()) -> non_neg_integer()) | atom(), [any()]) -> 'ok'.
@@ -107,10 +107,11 @@
 -define(INFO_KEYS, [name, type, durable, auto_delete, internal, arguments,
                     policy, user_who_performed_action]).
 
-recover() ->
+recover(VHost) ->
     Xs = rabbit_misc:table_filter(
            fun (#exchange{name = XName}) ->
-                   mnesia:read({rabbit_exchange, XName}) =:= []
+                XName#resource.virtual_host =:= VHost andalso
+                mnesia:read({rabbit_exchange, XName}) =:= []
            end,
            fun (X, Tx) ->
                    X1 = case Tx of

src/rabbit_mirror_queue_master.erl

@@ -26,7 +26,7 @@
          msg_rates/1, info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
          zip_msgs_and_acks/4]).
 
--export([start/1, stop/0, delete_crashed/1]).
+-export([start/2, stop/1, delete_crashed/1]).
 
 -export([promote_backing_queue_state/8, sender_death_fun/0, depth_fun/0]).
 
@@ -81,12 +81,12 @@
 %% Backing queue
 %% ---------------------------------------------------------------------------
 
-start(_DurableQueues) ->
+start(_Vhost, _DurableQueues) ->
     %% This will never get called as this module will never be
     %% installed as the default BQ implementation.
     exit({not_valid_for_generic_backing_queue, ?MODULE}).
 
-stop() ->
+stop(_Vhost) ->
     %% Same as start/1.
     exit({not_valid_for_generic_backing_queue, ?MODULE}).
 

src/rabbit_mirror_queue_slave.erl

@@ -194,6 +194,7 @@ stop_pending_slaves(QName, Pids) ->
     [begin
          rabbit_mirror_queue_misc:log_warning(
            QName, "Detected stale HA slave, stopping it: ~p~n", [Pid]),
+        %TODO: per-vhost supervisor
          case erlang:process_info(Pid, dictionary) of
              undefined -> ok;
              {dictionary, Dict} ->

src/rabbit_msg_store_vhost_sup.erl

@@ -30,7 +30,7 @@
 
 -export([enable/0]).
 
--export([start/1, stop/0]).
+-export([start/2, stop/1]).
 
 -export([init/3, terminate/2, delete_and_terminate/2, delete_crashed/1,
          purge/1, purge_acks/1,
@@ -83,22 +83,22 @@ enable() ->
 
 %%----------------------------------------------------------------------------
 
-start(QNames) ->
+start(VHost, QNames) ->
     BQ = bq(),
     %% TODO this expand-collapse dance is a bit ridiculous but it's what
     %% rabbit_amqqueue:recover/0 expects. We could probably simplify
     %% this if we rejigged recovery a bit.
     {DupNames, ExpNames} = expand_queues(QNames),
-    case BQ:start(ExpNames) of
+    case BQ:start(VHost, ExpNames) of
         {ok, ExpRecovery} ->
             {ok, collapse_recovery(QNames, DupNames, ExpRecovery)};
         Else ->
             Else
     end.
 
-stop() ->
+stop(VHost) ->
     BQ = bq(),
-    BQ:stop().
+    BQ:stop(VHost).
 
 %%----------------------------------------------------------------------------
 

src/rabbit_priority_queue.erl

@@ -17,10 +17,10 @@
 -module(rabbit_queue_index).
 
 -export([erase/1, init/3, reset_state/1, recover/6,
-         terminate/2, delete_and_terminate/1,
+         terminate/3, delete_and_terminate/1,
          pre_publish/7, flush_pre_publish_cache/2,
          publish/6, deliver/2, ack/2, sync/1, needs_sync/1, flush/1,
-         read/3, next_segment_boundary/1, bounds/1, start/1, stop/0]).
+         read/3, next_segment_boundary/1, bounds/1, start/2, stop/1]).
 
 -export([add_queue_ttl/0, avoid_zeroes/0, store_msg_size/0, store_msg/0]).
 -export([scan_queue_segments/3]).
@@ -261,7 +261,7 @@
                     on_sync_fun(), on_sync_fun()) ->
                         {'undefined' | non_neg_integer(),
                          'undefined' | non_neg_integer(), qistate()}.
--spec terminate([any()], qistate()) -> qistate().
+-spec terminate(rabbit_types:vhsot(), [any()], qistate()) -> qistate().
 -spec delete_and_terminate(qistate()) -> qistate().
 -spec publish(rabbit_types:msg_id(), seq_id(),
                     rabbit_types:message_properties(), boolean(),
@@ -278,7 +278,7 @@
 -spec next_segment_boundary(seq_id()) -> seq_id().
 -spec bounds(qistate()) ->
                        {non_neg_integer(), non_neg_integer(), qistate()}.
--spec start([rabbit_amqqueue:name()]) -> {[[any()]], {walker(A), A}}.
+-spec start(rabbit_types:vhsot(), [rabbit_amqqueue:name()]) -> {[[any()]], {walker(A), A}}.
 
 -spec add_queue_ttl() -> 'ok'.
 
@@ -321,9 +321,9 @@ recover(Name, Terms, MsgStoreRecovered, ContainsCheckFun,
         false -> init_dirty(CleanShutdown, ContainsCheckFun, State1)
     end.
 
-terminate(Terms, State = #qistate { dir = Dir }) ->
+terminate(VHost, Terms, State = #qistate { dir = Dir }) ->
     {SegmentCounts, State1} = terminate(State),
-    rabbit_recovery_terms:store(filename:basename(Dir),
+    rabbit_recovery_terms:store(VHost, filename:basename(Dir),
                                 [{segments, SegmentCounts} | Terms]),
     State1.
 
@@ -491,34 +491,36 @@ bounds(State = #qistate { segments = Segments }) ->
         end,
     {LowSeqId, NextSeqId, State}.
 
-start(DurableQueueNames) ->
-    ok = rabbit_recovery_terms:start(),
+start(VHost, DurableQueueNames) ->
+    ok = rabbit_recovery_terms:start(VHost),
     {DurableTerms, DurableDirectories} =
         lists:foldl(
           fun(QName, {RecoveryTerms, ValidDirectories}) ->
                   DirName = queue_name_to_dir_name(QName),
-                  RecoveryInfo = case rabbit_recovery_terms:read(DirName) of
+                  RecoveryInfo = case rabbit_recovery_terms:read(VHost, DirName) of
                                      {error, _}  -> non_clean_shutdown;
                                      {ok, Terms} -> Terms
                                  end,
                   {[RecoveryInfo | RecoveryTerms],
                    sets:add_element(DirName, ValidDirectories)}
           end, {[], sets:new()}, DurableQueueNames),
-
     %% Any queue directory we've not been asked to recover is considered garbage
     rabbit_file:recursive_delete(
       [DirName ||
-        DirName <- all_queue_directory_names(),
+        DirName <- all_queue_directory_names(VHost),
         not sets:is_element(filename:basename(DirName), DurableDirectories)]),
-
-    rabbit_recovery_terms:clear(),
+    rabbit_recovery_terms:clear(VHost),
 
     %% The backing queue interface requires that the queue recovery terms
     %% which come back from start/1 are in the same order as DurableQueueNames
     OrderedTerms = lists:reverse(DurableTerms),
     {OrderedTerms, {fun queue_index_walker/1, {start, DurableQueueNames}}}.
 
-stop() -> rabbit_recovery_terms:stop().
+stop(VHost) -> rabbit_recovery_terms:stop(VHost).
+
+all_queue_directory_names(VHost) ->
+    filelib:wildcard(filename:join([rabbit_vhost:msg_store_dir_path(VHost),
+                                    "queues", "*"])).
 
 all_queue_directory_names() ->
     filelib:wildcard(filename:join([rabbit_vhost:msg_store_dir_wildcard(),
@@ -1447,6 +1449,6 @@ move_to_per_vhost_stores(#resource{} = QueueName) ->
     end,
     ok.
 
-update_recovery_term(#resource{} = QueueName, Term) ->
+update_recovery_term(#resource{virtual_host = VHost} = QueueName, Term) ->
     Key = queue_name_to_dir_name(QueueName),
-    rabbit_recovery_terms:store(Key, Term).
+    rabbit_recovery_terms:store(VHost, Key, Term).