Refuse connections to dead vhosts.

If a vhost supervision tree is not active, which can be
a result of an error in message store, refuse connections
to this vhost on the node.
This is a follow-up to [#140841611] [#1158]

[#145106713]

2 files changed, 47 insertions, 16 deletions

diff --git a/src/rabbit_direct.erl b/src/rabbit_direct.erl
index 4b7f06305a..26e8f4d452 100644
--- a/src/rabbit_direct.erl
+++ b/src/rabbit_direct.erl
@@ -90,16 +90,21 @@ connect(Creds, VHost, Protocol, Pid, Infos) ->
                 true  ->
                     {error, not_allowed};
                 false ->
-                    case AuthFun() of
-                        {ok, User = #user{username = Username}} ->
-                            notify_auth_result(Username,
-                                user_authentication_success, []),
-                            connect1(User, VHost, Protocol, Pid, Infos);
-                        {refused, Username, Msg, Args} ->
-                            notify_auth_result(Username,
-                                user_authentication_failure,
-                                [{error, rabbit_misc:format(Msg, Args)}]),
-                            {error, {auth_failure, "Refused"}}
+                    case is_vhost_alive(VHost, Creds, Pid) of
+                        false ->
+                            {error, {internal_error, vhost_is_down}};
+                        true  ->
+                            case AuthFun() of
+                                {ok, User = #user{username = Username}} ->
+                                    notify_auth_result(Username,
+                                        user_authentication_success, []),
+                                    connect1(User, VHost, Protocol, Pid, Infos);
+                                {refused, Username, Msg, Args} ->
+                                    notify_auth_result(Username,
+                                        user_authentication_failure,
+                                        [{error, rabbit_misc:format(Msg, Args)}]),
+                                    {error, {auth_failure, "Refused"}}
+                            end
                     end
             end;
         false -> {error, broker_not_found_on_node}
@@ -140,6 +145,21 @@ maybe_call_connection_info_module(Protocol, Creds, VHost, Pid, Infos) ->
             []
     end.
 
+is_vhost_alive(VHost, {Username, _Password}, Pid) ->
+    PrintedUsername = case Username of
+        none -> "";
+        _    -> Username
+    end,
+    case rabbit_vhost_sup_sup:is_vhost_alive(VHost) of
+        true  -> true;
+        false ->
+            rabbit_log_connection:error(
+                "Error on Direct connection ~p~n"
+                "access to vhost '~s' refused for user '~s': "
+                "vhost '~s' is down",
+                [Pid, VHost, PrintedUsername, VHost]),
+            false
+    end.
 
 is_over_connection_limit(VHost, {Username, _Password}, Pid) ->
     PrintedUsername = case Username of
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index e23d382d6e..77914a00bf 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -567,7 +567,7 @@ handle_other(handshake_timeout, State) ->
     throw({handshake_timeout, State#v1.callback});
 handle_other(heartbeat_timeout, State = #v1{connection_state = closed}) ->
     State;
-handle_other(heartbeat_timeout, 
+handle_other(heartbeat_timeout,
              State = #v1{connection = #connection{timeout_sec = T}}) ->
     maybe_emit_stats(State),
     throw({heartbeat_timeout, T});
@@ -623,7 +623,7 @@ send_blocked(#v1{connection = #connection{protocol     = Protocol,
                  sock       = Sock}, Reason) ->
     case rabbit_misc:table_lookup(Capabilities, <<"connection.blocked">>) of
         {bool, true} ->
-            
+
             ok = send_on_channel0(Sock, #'connection.blocked'{reason = Reason},
                                   Protocol);
         _ ->
@@ -1164,6 +1164,7 @@ handle_method0(#'connection.open'{virtual_host = VHost},
 
     ok = is_over_connection_limit(VHost, User),
     ok = rabbit_access_control:check_vhost_access(User, VHost, Sock),
+    ok = is_vhost_alive(VHost, User),
     NewConnection = Connection#connection{vhost = VHost},
     ok = send_on_channel0(Sock, #'connection.open_ok'{}, Protocol),
 
@@ -1209,6 +1210,16 @@ handle_method0(_Method, #v1{connection_state = S}) ->
     rabbit_misc:protocol_error(
       channel_error, "unexpected method in connection state ~w", [S]).
 
+is_vhost_alive(VHostPath, User) ->
+    case rabbit_vhost_sup_sup:is_vhost_alive(VHostPath) of
+        true  -> ok;
+        false ->
+            rabbit_misc:protocol_error(internal_error,
+                            "access to vhost '~s' refused for user '~s': "
+                            "vhost '~s' is down",
+                            [VHostPath, User#user.username, VHostPath])
+    end.
+
 is_over_connection_limit(VHostPath, User) ->
     try rabbit_vhost_limit:is_over_connection_limit(VHostPath) of
         false         -> ok;
@@ -1567,7 +1578,7 @@ maybe_block(State = #v1{connection_state = CS, throttle = Throttle}) ->
             State1 = State#v1{connection_state = blocked,
                               throttle = update_last_blocked_at(Throttle)},
             case CS of
-                running -> 
+                running ->
                     ok = rabbit_heartbeat:pause_monitor(State#v1.heartbeater);
                 _       -> ok
             end,
@@ -1589,7 +1600,7 @@ maybe_send_unblocked(State = #v1{throttle = Throttle}) ->
     case should_send_unblocked(Throttle) of
         true ->
             ok = send_unblocked(State),
-            State#v1{throttle = 
+            State#v1{throttle =
                 Throttle#throttle{connection_blocked_message_sent = false}};
         false -> State
     end.
@@ -1598,7 +1609,7 @@ maybe_send_blocked_or_unblocked(State = #v1{throttle = Throttle}) ->
     case should_send_blocked(Throttle) of
         true ->
             ok = send_blocked(State, blocked_by_message(Throttle)),
-            State#v1{throttle = 
+            State#v1{throttle =
                 Throttle#throttle{connection_blocked_message_sent = true}};
         false -> maybe_send_unblocked(State)
     end.
@@ -1624,7 +1635,7 @@ control_throttle(State = #v1{connection_state = CS,
         running -> maybe_block(State1);
         %% unblock or re-enable blocking
         blocked -> maybe_block(maybe_unblock(State1));
-        _       -> State1 
+        _       -> State1
     end.
 
 augment_connection_log_name(#connection{client_properties = ClientProperties,