Merge bug26427HEADmaster

33 files changed, 1133 insertions, 262 deletions

diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index 8d04f28a..8d042670 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -426,6 +426,41 @@
           </listitem>
         </varlistentry>
         <varlistentry>
+          <term><cmdsynopsis><command>rename_cluster_node</command> <arg choice="req">oldnode1</arg> <arg choice="req">newnode1</arg> <arg choice="opt">oldnode2</arg> <arg choice="opt">newnode2 ...</arg></cmdsynopsis></term>
+          <listitem>
+            <para>
+              Supports renaming of cluster nodes in the local database.
+            </para>
+            <para>
+              This subcommand causes rabbitmqctl to temporarily become
+              the node in order to make the change. The local cluster
+              node must therefore be completely stopped; other nodes
+              can be online or offline.
+            </para>
+            <para>
+              This subcommand takes an even number of arguments, in
+              pairs representing the old and new names for nodes. You
+              must specify the old and new names for this node and for
+              any other nodes that are stopped and being renamed at
+              the same time.
+            </para>
+            <para>
+              It is possible to stop all nodes and rename them all
+              simultaneously (in which case old and new names for all
+              nodes must be given to every node) or stop and rename
+              nodes one at a time (in which case each node only needs
+              to be told how its own name is changing).
+            </para>
+            <para role="example-prefix">For example:</para>
+            <screen role="example">rabbitmqctl rename_cluster_node rabbit@misshelpful rabbit@cordelia</screen>
+            <para role="example">
+              This command will rename the node
+              <command>rabbit@misshelpful</command> to the node
+              <command>rabbit@cordelia</command>.
+            </para>
+          </listitem>
+        </varlistentry>
+        <varlistentry>
           <term><cmdsynopsis><command>update_cluster_nodes</command> <arg choice="req">clusternode</arg></cmdsynopsis>
           </term>
           <listitem>
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 74e165cd..9cbd978e 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -14,12 +14,17 @@
 %% Copyright (c) 2007-2014 GoPivotal, Inc.  All rights reserved.
 %%
 
+%% Passed around most places
 -record(user, {username,
                tags,
-               auth_backend, %% Module this user came from
-               impl          %% Scratch space for that module
-              }).
+               authz_backends}). %% List of {Module, AuthUserImpl} pairs
 
+%% Passed to auth backends
+-record(auth_user, {username,
+                    tags,
+                    impl}).
+
+%% Implementation for the internal auth backend
 -record(internal_user, {username, password_hash, tags}).
 -record(permission, {configure, write, read}).
 -record(user_vhost, {username, virtual_host}).
diff --git a/packaging/RPMS/Fedora/rabbitmq-server.spec b/packaging/RPMS/Fedora/rabbitmq-server.spec
index c77d7e9d..60a81914 100644
--- a/packaging/RPMS/Fedora/rabbitmq-server.spec
+++ b/packaging/RPMS/Fedora/rabbitmq-server.spec
@@ -130,6 +130,9 @@ done
 rm -rf %{buildroot}
 
 %changelog
+* Tue Jan 6 2015 jean-sebastien@rabbitmq.com 3.4.3-1
+- New Upstream Release
+
 * Wed Nov 26 2014 simon@rabbitmq.com 3.4.2-1
 - New Upstream Release
 
diff --git a/packaging/debs/Debian/debian/changelog b/packaging/debs/Debian/debian/changelog
index 5e3744fd..47ec527c 100644
--- a/packaging/debs/Debian/debian/changelog
+++ b/packaging/debs/Debian/debian/changelog
@@ -1,3 +1,9 @@
+rabbitmq-server (3.4.3-1) unstable; urgency=low
+
+  * New Upstream Release
+
+ -- Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com>  Tue, 06 Jan 2015 15:58:45 +0000
+
 rabbitmq-server (3.4.2-1) unstable; urgency=low
 
   * New Upstream Release
diff --git a/packaging/windows/Makefile b/packaging/windows/Makefile
index 53fc31fc..5dc802a8 100644
--- a/packaging/windows/Makefile
+++ b/packaging/windows/Makefile
@@ -7,9 +7,9 @@ dist:
 	tar -zxf ../../dist/$(SOURCE_DIR).tar.gz
 	$(MAKE) -C $(SOURCE_DIR)
 
-	mkdir $(SOURCE_DIR)/sbin
+	mkdir -p $(SOURCE_DIR)/sbin
 	mv $(SOURCE_DIR)/scripts/*.bat $(SOURCE_DIR)/sbin
-	mkdir $(SOURCE_DIR)/etc
+	mkdir -p $(SOURCE_DIR)/etc
 	cp $(SOURCE_DIR)/docs/rabbitmq.config.example $(SOURCE_DIR)/etc/rabbitmq.config.example
 	cp README-etc $(SOURCE_DIR)/etc/README.txt
 	rm -rf $(SOURCE_DIR)/scripts
diff --git a/scripts/rabbitmq-env b/scripts/rabbitmq-env
index 63cfda3c..5a3e73bc 100755
--- a/scripts/rabbitmq-env
+++ b/scripts/rabbitmq-env
@@ -111,3 +111,6 @@ DEFAULT_NODE_PORT=5672
 [ "x" = "x$RABBITMQ_CTL_ERL_ARGS" ] && RABBITMQ_CTL_ERL_ARGS=${CTL_ERL_ARGS}
 
 ##--- End of overridden <var_name> variables
+
+# Since we source this elsewhere, don't accidentally stop execution
+true
diff --git a/src/file_handle_cache.erl b/src/file_handle_cache.erl
index 3a7a692c..2922e146 100644
--- a/src/file_handle_cache.erl
+++ b/src/file_handle_cache.erl
@@ -178,6 +178,9 @@
           write_buffer_size,
           write_buffer_size_limit,
           write_buffer,
+          read_buffer,
+          read_buffer_size,
+          read_buffer_size_limit,
           at_eof,
           path,
           mode,
@@ -237,7 +240,8 @@
 -spec(register_callback/3 :: (atom(), atom(), [any()]) -> 'ok').
 -spec(open/3 ::
         (file:filename(), [any()],
-         [{'write_buffer', (non_neg_integer() | 'infinity' | 'unbuffered')}])
+         [{'write_buffer', (non_neg_integer() | 'infinity' | 'unbuffered')} |
+          {'read_buffer', (non_neg_integer() | 'unbuffered')}])
         -> val_or_error(ref())).
 -spec(close/1 :: (ref()) -> ok_or_error()).
 -spec(read/2 :: (ref(), non_neg_integer()) ->
@@ -331,16 +335,44 @@ close(Ref) ->
 
 read(Ref, Count) ->
     with_flushed_handles(
-      [Ref],
+      [Ref], keep,
       fun ([#handle { is_read = false }]) ->
               {error, not_open_for_reading};
-          ([Handle = #handle { hdl = Hdl, offset = Offset }]) ->
-              case prim_file:read(Hdl, Count) of
-                  {ok, Data} = Obj -> Offset1 = Offset + iolist_size(Data),
-                                      {Obj,
-                                       [Handle #handle { offset = Offset1 }]};
-                  eof              -> {eof, [Handle #handle { at_eof = true }]};
-                  Error            -> {Error, [Handle]}
+          ([Handle = #handle{read_buffer      = Buf,
+                             read_buffer_size = BufSz,
+                             offset           = Offset}]) when BufSz >= Count ->
+              <<Hd:Count/binary, Tl/binary>> = Buf,
+              {{ok, Hd}, [Handle#handle{offset           = Offset + Count,
+                                        read_buffer      = Tl,
+                                        read_buffer_size = BufSz - Count}]};
+          ([Handle = #handle{read_buffer            = Buf,
+                             read_buffer_size       = BufSz,
+                             read_buffer_size_limit = BufSzLimit,
+                             hdl                    = Hdl,
+                             offset                 = Offset}]) ->
+              WantedCount = Count - BufSz,
+              case prim_file_read(Hdl, lists:max([BufSzLimit, WantedCount])) of
+                  {ok, Data} ->
+                      ReadCount = size(Data),
+                      case ReadCount < WantedCount of
+                          true ->
+                              OffSet1 = Offset + BufSz + ReadCount,
+                              {{ok, <<Buf/binary, Data/binary>>},
+                               [reset_read_buffer(
+                                  Handle#handle{offset = OffSet1})]};
+                          false ->
+                              <<Hd:WantedCount/binary, Tl/binary>> = Data,
+                              OffSet1 = Offset + BufSz + WantedCount,
+                              BufSz1 = ReadCount - WantedCount,
+                              {{ok, <<Buf/binary, Hd/binary>>},
+                               [Handle#handle{offset           = OffSet1,
+                                              read_buffer      = Tl,
+                                              read_buffer_size = BufSz1}]}
+                      end;
+                  eof ->
+                      {eof, [Handle #handle { at_eof = true }]};
+                  Error ->
+                      {Error, [reset_read_buffer(Handle)]}
               end
       end).
 
@@ -355,7 +387,7 @@ append(Ref, Data) ->
                                             write_buffer_size_limit = 0,
                                             at_eof = true } = Handle1} ->
                       Offset1 = Offset + iolist_size(Data),
-                      {prim_file:write(Hdl, Data),
+                      {prim_file_write(Hdl, Data),
                        [Handle1 #handle { is_dirty = true, offset = Offset1 }]};
                   {{ok, _Offset}, #handle { write_buffer = WriteBuffer,
                                             write_buffer_size = Size,
@@ -377,12 +409,12 @@ append(Ref, Data) ->
 
 sync(Ref) ->
     with_flushed_handles(
-      [Ref],
+      [Ref], keep,
       fun ([#handle { is_dirty = false, write_buffer = [] }]) ->
               ok;
           ([Handle = #handle { hdl = Hdl,
                                is_dirty = true, write_buffer = [] }]) ->
-              case prim_file:sync(Hdl) of
+              case prim_file_sync(Hdl) of
                   ok    -> {ok, [Handle #handle { is_dirty = false }]};
                   Error -> {Error, [Handle]}
               end
@@ -397,7 +429,7 @@ needs_sync(Ref) ->
 
 position(Ref, NewOffset) ->
     with_flushed_handles(
-      [Ref],
+      [Ref], keep,
       fun ([Handle]) -> {Result, Handle1} = maybe_seek(NewOffset, Handle),
                         {Result, [Handle1]}
       end).
@@ -465,8 +497,8 @@ clear(Ref) ->
       fun ([#handle { at_eof = true, write_buffer_size = 0, offset = 0 }]) ->
               ok;
           ([Handle]) ->
-              case maybe_seek(bof, Handle #handle { write_buffer = [],
-                                                    write_buffer_size = 0 }) of
+              case maybe_seek(bof, Handle#handle{write_buffer      = [],
+                                                 write_buffer_size = 0}) of
                   {{ok, 0}, Handle1 = #handle { hdl = Hdl }} ->
                       case prim_file:truncate(Hdl) of
                           ok    -> {ok, [Handle1 #handle { at_eof = true }]};
@@ -539,6 +571,21 @@ info(Items) -> gen_server2:call(?SERVER, {info, Items}, infinity).
 %% Internal functions
 %%----------------------------------------------------------------------------
 
+prim_file_read(Hdl, Size) ->
+    file_handle_cache_stats:update(
+      read, Size, fun() -> prim_file:read(Hdl, Size) end).
+
+prim_file_write(Hdl, Bytes) ->
+    file_handle_cache_stats:update(
+      write, iolist_size(Bytes), fun() -> prim_file:write(Hdl, Bytes) end).
+
+prim_file_sync(Hdl) ->
+    file_handle_cache_stats:update(sync, fun() -> prim_file:sync(Hdl) end).
+
+prim_file_position(Hdl, NewOffset) ->
+    file_handle_cache_stats:update(
+      seek, fun() -> prim_file:position(Hdl, NewOffset) end).
+
 is_reader(Mode) -> lists:member(read, Mode).
 
 is_writer(Mode) -> lists:member(write, Mode).
@@ -550,8 +597,15 @@ append_to_write(Mode) ->
     end.
 
 with_handles(Refs, Fun) ->
+    with_handles(Refs, reset, Fun).
+
+with_handles(Refs, ReadBuffer, Fun) ->
     case get_or_reopen([{Ref, reopen} || Ref <- Refs]) of
-        {ok, Handles} ->
+        {ok, Handles0} ->
+            Handles = case ReadBuffer of
+                          reset -> [reset_read_buffer(H) || H <- Handles0];
+                          keep  -> Handles0
+                      end,
             case Fun(Handles) of
                 {Result, Handles1} when is_list(Handles1) ->
                     lists:zipwith(fun put_handle/2, Refs, Handles1),
@@ -564,8 +618,11 @@ with_handles(Refs, Fun) ->
     end.
 
 with_flushed_handles(Refs, Fun) ->
+    with_flushed_handles(Refs, reset, Fun).
+
+with_flushed_handles(Refs, ReadBuffer, Fun) ->
     with_handles(
-      Refs,
+      Refs, ReadBuffer,
       fun (Handles) ->
               case lists:foldl(
                      fun (Handle, {ok, HandlesAcc}) ->
@@ -611,20 +668,23 @@ reopen([], Tree, RefHdls) ->
     {ok, lists:reverse(RefHdls)};
 reopen([{Ref, NewOrReopen, Handle = #handle { hdl          = closed,
                                               path         = Path,
-                                              mode         = Mode,
+                                              mode         = Mode0,
                                               offset       = Offset,
                                               last_used_at = undefined }} |
         RefNewOrReopenHdls] = ToOpen, Tree, RefHdls) ->
-    case prim_file:open(Path, case NewOrReopen of
-                                  new    -> Mode;
-                                  reopen -> [read | Mode]
-                              end) of
+    Mode = case NewOrReopen of
+               new    -> Mode0;
+               reopen -> file_handle_cache_stats:update(reopen),
+                         [read | Mode0]
+           end,
+    case prim_file:open(Path, Mode) of
         {ok, Hdl} ->
             Now = now(),
             {{ok, _Offset}, Handle1} =
-                maybe_seek(Offset, Handle #handle { hdl          = Hdl,
-                                                    offset       = 0,
-                                                    last_used_at = Now }),
+                maybe_seek(Offset, reset_read_buffer(
+                                     Handle#handle{hdl              = Hdl,
+                                                   offset           = 0,
+                                                   last_used_at     = Now})),
             put({Ref, fhc_handle}, Handle1),
             reopen(RefNewOrReopenHdls, gb_trees:insert(Now, Ref, Tree),
                    [{Ref, Handle1} | RefHdls]);
@@ -709,6 +769,11 @@ new_closed_handle(Path, Mode, Options) ->
             infinity             -> infinity;
             N when is_integer(N) -> N
         end,
+    ReadBufferSize =
+        case proplists:get_value(read_buffer, Options, unbuffered) of
+            unbuffered             -> 0;
+            N2 when is_integer(N2) -> N2
+        end,
     Ref = make_ref(),
     put({Ref, fhc_handle}, #handle { hdl                     = closed,
                                      offset                  = 0,
@@ -716,6 +781,9 @@ new_closed_handle(Path, Mode, Options) ->
                                      write_buffer_size       = 0,
                                      write_buffer_size_limit = WriteBufferSize,
                                      write_buffer            = [],
+                                     read_buffer_size        = 0,
+                                     read_buffer_size_limit  = ReadBufferSize,
+                                     read_buffer             = <<>>,
                                      at_eof                  = false,
                                      path                    = Path,
                                      mode                    = Mode,
@@ -742,7 +810,7 @@ soft_close(Handle) ->
                        is_dirty    = IsDirty,
                        last_used_at = Then } = Handle1 } ->
             ok = case IsDirty of
-                     true  -> prim_file:sync(Hdl);
+                     true  -> prim_file_sync(Hdl);
                      false -> ok
                  end,
             ok = prim_file:close(Hdl),
@@ -776,17 +844,31 @@ hard_close(Handle) ->
             Result
     end.
 
-maybe_seek(NewOffset, Handle = #handle { hdl = Hdl, offset = Offset,
-                                         at_eof = AtEoF }) ->
+maybe_seek(NewOffset, Handle = #handle{hdl              = Hdl,
+                                       offset           = Offset,
+                                       read_buffer      = Buf,
+                                       read_buffer_size = BufSz,
+                                       at_eof           = AtEoF}) ->
     {AtEoF1, NeedsSeek} = needs_seek(AtEoF, Offset, NewOffset),
-    case (case NeedsSeek of
-              true  -> prim_file:position(Hdl, NewOffset);
-              false -> {ok, Offset}
-          end) of
-        {ok, Offset1} = Result ->
-            {Result, Handle #handle { offset = Offset1, at_eof = AtEoF1 }};
-        {error, _} = Error ->
-            {Error, Handle}
+    case NeedsSeek of
+        true when is_number(NewOffset) andalso
+                  NewOffset >= Offset andalso NewOffset =< BufSz + Offset ->
+            Diff = NewOffset - Offset,
+            <<_:Diff/binary, Rest/binary>> = Buf,
+            {{ok, NewOffset}, Handle#handle{offset           = NewOffset,
+                                            at_eof           = AtEoF1,
+                                            read_buffer      = Rest,
+                                            read_buffer_size = BufSz - Diff}};
+        true ->
+            case prim_file_position(Hdl, NewOffset) of
+                {ok, Offset1} = Result ->
+                    {Result, reset_read_buffer(Handle#handle{offset = Offset1,
+                                                             at_eof = AtEoF1})};
+                {error, _} = Error ->
+                    {Error, Handle}
+            end;
+        false ->
+            {{ok, Offset}, Handle}
     end.
 
 needs_seek( AtEoF, _CurOffset,  cur     ) -> {AtEoF, false};
@@ -817,7 +899,7 @@ write_buffer(Handle = #handle { hdl = Hdl, offset = Offset,
                                 write_buffer = WriteBuffer,
                                 write_buffer_size = DataSize,
                                 at_eof = true }) ->
-    case prim_file:write(Hdl, lists:reverse(WriteBuffer)) of
+    case prim_file_write(Hdl, lists:reverse(WriteBuffer)) of
         ok ->
             Offset1 = Offset + DataSize,
             {ok, Handle #handle { offset = Offset1, is_dirty = true,
@@ -826,6 +908,10 @@ write_buffer(Handle = #handle { hdl = Hdl, offset = Offset,
             {Error, Handle}
     end.
 
+reset_read_buffer(Handle) ->
+    Handle#handle{read_buffer      = <<>>,
+                  read_buffer_size = 0}.
+
 infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
 
 i(total_limit,   #fhc_state{limit               = Limit}) -> Limit;
@@ -843,6 +929,7 @@ used(#fhc_state{open_count          = C1,
 %%----------------------------------------------------------------------------
 
 init([AlarmSet, AlarmClear]) ->
+    file_handle_cache_stats:init(),
     Limit = case application:get_env(file_handles_high_watermark) of
                 {ok, Watermark} when (is_integer(Watermark) andalso
                                       Watermark > 0) ->
diff --git a/src/file_handle_cache_stats.erl b/src/file_handle_cache_stats.erl
new file mode 100644
index 00000000..832f0b3d
--- /dev/null
+++ b/src/file_handle_cache_stats.erl
@@ -0,0 +1,60 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2014 GoPivotal, Inc.  All rights reserved.
+%%
+
+-module(file_handle_cache_stats).
+
+%% stats about read / write operations that go through the fhc.
+
+-export([init/0, update/3, update/2, update/1, get/0]).
+
+-define(TABLE, ?MODULE).
+
+init() ->
+    ets:new(?TABLE, [public, named_table]),
+    [ets:insert(?TABLE, {{Op, Counter}, 0}) || Op      <- [read, write],
+                                               Counter <- [count, bytes, time]],
+    [ets:insert(?TABLE, {{Op, Counter}, 0}) || Op      <- [sync, seek],
+                                               Counter <- [count, time]],
+    [ets:insert(?TABLE, {{Op, Counter}, 0}) || Op      <- [reopen],
+                                               Counter <- [count]].
+
+update(Op, Bytes, Thunk) ->
+    {Time, Res} = timer_tc(Thunk),
+    ets:update_counter(?TABLE, {Op, count}, 1),
+    ets:update_counter(?TABLE, {Op, bytes}, Bytes),
+    ets:update_counter(?TABLE, {Op, time}, Time),
+    Res.
+
+update(Op, Thunk) ->
+    {Time, Res} = timer_tc(Thunk),
+    ets:update_counter(?TABLE, {Op, count}, 1),
+    ets:update_counter(?TABLE, {Op, time}, Time),
+    Res.
+
+update(Op) ->
+    ets:update_counter(?TABLE, {Op, count}, 1),
+    ok.
+
+get() ->
+    lists:sort(ets:tab2list(?TABLE)).
+
+%% TODO timer:tc/1 was introduced in R14B03; use that function once we
+%% require that version.
+timer_tc(Thunk) ->
+    T1 = os:timestamp(),
+    Res = Thunk(),
+    T2 = os:timestamp(),
+    {timer:now_diff(T2, T1), Res}.
diff --git a/src/rabbit.erl b/src/rabbit.erl
index 664da206..40f24efc 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -243,15 +243,19 @@ maybe_hipe_compile() ->
     {ok, Want} = application:get_env(rabbit, hipe_compile),
     Can = code:which(hipe) =/= non_existing,
     case {Want, Can} of
-        {true,  true}  -> hipe_compile(),
-                          true;
+        {true,  true}  -> hipe_compile();
         {true,  false} -> false;
-        {false, _}     -> true
+        {false, _}     -> {ok, disabled}
     end.
 
-warn_if_hipe_compilation_failed(true) ->
+log_hipe_result({ok, disabled}) ->
     ok;
-warn_if_hipe_compilation_failed(false) ->
+log_hipe_result({ok, Count, Duration}) ->
+    rabbit_log:info(
+      "HiPE in use: compiled ~B modules in ~Bs.~n", [Count, Duration]);
+log_hipe_result(false) ->
+    io:format(
+      "~nNot HiPE compiling: HiPE not found in this Erlang installation.~n"),
     rabbit_log:warning(
       "Not HiPE compiling: HiPE not found in this Erlang installation.~n").
 
@@ -276,8 +280,9 @@ hipe_compile() ->
          {'DOWN', MRef, process, _, Reason} -> exit(Reason)
      end || {_Pid, MRef} <- PidMRefs],
     T2 = erlang:now(),
-    io:format("|~n~nCompiled ~B modules in ~Bs~n",
-              [Count, timer:now_diff(T2, T1) div 1000000]).
+    Duration = timer:now_diff(T2, T1) div 1000000,
+    io:format("|~n~nCompiled ~B modules in ~Bs~n", [Count, Duration]),
+    {ok, Count, Duration}.
 
 split(L, N) -> split0(L, [[] || _ <- lists:seq(1, N)]).
 
@@ -307,9 +312,9 @@ start() ->
 boot() ->
     start_it(fun() ->
                      ok = ensure_application_loaded(),
-                     Success = maybe_hipe_compile(),
+                     HipeResult = maybe_hipe_compile(),
                      ok = ensure_working_log_handlers(),
-                     warn_if_hipe_compilation_failed(Success),
+                     log_hipe_result(HipeResult),
                      rabbit_node_monitor:prepare_cluster_status_files(),
                      ok = rabbit_upgrade:maybe_upgrade_mnesia(),
                      %% It's important that the consistency check happens after
diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index b0a9c0d8..41c54b07 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -19,7 +19,7 @@
 -include("rabbit.hrl").
 
 -export([check_user_pass_login/2, check_user_login/2, check_user_loopback/2,
-         check_vhost_access/2, check_resource_access/3]).
+         check_vhost_access/3, check_resource_access/3]).
 
 %%----------------------------------------------------------------------------
 
@@ -31,15 +31,17 @@
 
 -spec(check_user_pass_login/2 ::
         (rabbit_types:username(), rabbit_types:password())
-        -> {'ok', rabbit_types:user()} | {'refused', string(), [any()]}).
+        -> {'ok', rabbit_types:user()} |
+           {'refused', rabbit_types:username(), string(), [any()]}).
 -spec(check_user_login/2 ::
         (rabbit_types:username(), [{atom(), any()}])
-        -> {'ok', rabbit_types:user()} | {'refused', string(), [any()]}).
+        -> {'ok', rabbit_types:user()} |
+           {'refused', rabbit_types:username(), string(), [any()]}).
 -spec(check_user_loopback/2 :: (rabbit_types:username(),
                                 rabbit_net:socket() | inet:ip_address())
         -> 'ok' | 'not_allowed').
--spec(check_vhost_access/2 ::
-        (rabbit_types:user(), rabbit_types:vhost())
+-spec(check_vhost_access/3 ::
+        (rabbit_types:user(), rabbit_types:vhost(), rabbit_net:socket())
         -> 'ok' | rabbit_types:channel_exit()).
 -spec(check_resource_access/3 ::
         (rabbit_types:user(), rabbit_types:r(atom()), permission_atom())
@@ -55,36 +57,71 @@ check_user_pass_login(Username, Password) ->
 check_user_login(Username, AuthProps) ->
     {ok, Modules} = application:get_env(rabbit, auth_backends),
     R = lists:foldl(
-          fun ({ModN, ModZ}, {refused, _, _}) ->
+          fun ({ModN, ModZs0}, {refused, _, _, _}) ->
+                  ModZs = case ModZs0 of
+                              A when is_atom(A) -> [A];
+                              L when is_list(L) -> L
+                          end,
                   %% Different modules for authN vs authZ. So authenticate
                   %% with authN module, then if that succeeds do
-                  %% passwordless (i.e pre-authenticated) login with authZ
-                  %% module, and use the #user{} the latter gives us.
-                  case try_login(ModN, Username, AuthProps) of
-                      {ok, _} -> try_login(ModZ, Username, []);
-                      Else    -> Else
+                  %% passwordless (i.e pre-authenticated) login with authZ.
+                  case try_authenticate(ModN, Username, AuthProps) of
+                      {ok, ModNUser = #auth_user{username = Username2}} ->
+                          user(ModNUser, try_authorize(ModZs, Username2));
+                      Else ->
+                          Else
                   end;
-              (Mod, {refused, _, _}) ->
+              (Mod, {refused, _, _, _}) ->
                   %% Same module for authN and authZ. Just take the result
                   %% it gives us
-                  try_login(Mod, Username, AuthProps);
+                  case try_authenticate(Mod, Username, AuthProps) of
+                      {ok, ModNUser = #auth_user{impl = Impl}} ->
+                          user(ModNUser, {ok, [{Mod, Impl}]});
+                      Else ->
+                          Else
+                  end;
               (_, {ok, User}) ->
                   %% We've successfully authenticated. Skip to the end...
                   {ok, User}
-          end, {refused, "No modules checked '~s'", [Username]}, Modules),
-    rabbit_event:notify(case R of
-                            {ok, _User} -> user_authentication_success;
-                            _           -> user_authentication_failure
-                        end, [{name, Username}]),
+          end,
+          {refused, Username, "No modules checked '~s'", [Username]}, Modules),
     R.
 
-try_login(Module, Username, AuthProps) ->
-    case Module:check_user_login(Username, AuthProps) of
-        {error, E} -> {refused, "~s failed authenticating ~s: ~p~n",
-                       [Module, Username, E]};
-        Else       -> Else
+try_authenticate(Module, Username, AuthProps) ->
+    case Module:user_login_authentication(Username, AuthProps) of
+        {ok, AuthUser}  -> {ok, AuthUser};
+        {error, E}      -> {refused, Username,
+                            "~s failed authenticating ~s: ~p~n",
+                            [Module, Username, E]};
+        {refused, F, A} -> {refused, Username, F, A}
     end.
 
+try_authorize(Modules, Username) ->
+    lists:foldr(
+      fun (Module, {ok, ModsImpls}) ->
+              case Module:user_login_authorization(Username) of
+                  {ok, Impl}      -> {ok, [{Module, Impl} | ModsImpls]};
+                  {error, E}      -> {refused, Username,
+                                        "~s failed authorizing ~s: ~p~n",
+                                        [Module, Username, E]};
+                  {refused, F, A} -> {refused, Username, F, A}
+              end;
+          (_,      {refused, F, A}) ->
+              {refused, Username, F, A}
+      end, {ok, []}, Modules).
+
+user(#auth_user{username = Username, tags = Tags}, {ok, ModZImpls}) ->
+    {ok, #user{username       = Username,
+               tags           = Tags,
+               authz_backends = ModZImpls}};
+user(_AuthUser, Error) ->
+    Error.
+
+auth_user(#user{username = Username, tags = Tags}, Impl) ->
+    #auth_user{username = Username,
+               tags     = Tags,
+               impl     = Impl}.
+
 check_user_loopback(Username, SockOrAddr) ->
     {ok, Users} = application:get_env(rabbit, loopback_users),
     case rabbit_net:is_loopback(SockOrAddr)
@@ -93,29 +130,38 @@ check_user_loopback(Username, SockOrAddr) ->
         false -> not_allowed
     end.
 
-check_vhost_access(User = #user{ username     = Username,
-                                 auth_backend = Module }, VHostPath) ->
-    check_access(
-      fun() ->
-              %% TODO this could be an andalso shortcut under >R13A
-              case rabbit_vhost:exists(VHostPath) of
-                  false -> false;
-                  true  -> Module:check_vhost_access(User, VHostPath)
-              end
-      end,
-      Module, "access to vhost '~s' refused for user '~s'",
-      [VHostPath, Username]).
+check_vhost_access(User = #user{username       = Username,
+                                authz_backends = Modules}, VHostPath, Sock) ->
+    lists:foldl(
+      fun({Mod, Impl}, ok) ->
+              check_access(
+                fun() ->
+                        rabbit_vhost:exists(VHostPath) andalso
+                            Mod:check_vhost_access(
+                              auth_user(User, Impl), VHostPath, Sock)
+                end,
+                Mod, "access to vhost '~s' refused for user '~s'",
+                [VHostPath, Username]);
+         (_, Else) ->
+              Else
+      end, ok, Modules).
 
 check_resource_access(User, R = #resource{kind = exchange, name = <<"">>},
                       Permission) ->
     check_resource_access(User, R#resource{name = <<"amq.default">>},
                           Permission);
-check_resource_access(User = #user{username = Username, auth_backend = Module},
+check_resource_access(User = #user{username       = Username,
+                                   authz_backends = Modules},
                       Resource, Permission) ->
-    check_access(
-      fun() -> Module:check_resource_access(User, Resource, Permission) end,
-      Module, "access to ~s refused for user '~s'",
-      [rabbit_misc:rs(Resource), Username]).
+    lists:foldl(
+      fun({Module, Impl}, ok) ->
+              check_access(
+                fun() -> Module:check_resource_access(
+                           auth_user(User, Impl), Resource, Permission) end,
+                Module, "access to ~s refused for user '~s'",
+                [rabbit_misc:rs(Resource), Username]);
+         (_, Else) -> Else
+      end, ok, Modules).
 
 check_access(Fun, Module, ErrStr, ErrArgs) ->
     Allow = case Fun() of
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index b06da4c1..a18df225 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -84,7 +84,6 @@
          slave_pids,
          synchronised_slave_pids,
          down_slave_nodes,
-         backing_queue_status,
          state
         ]).
 
diff --git a/src/rabbit_auth_backend_dummy.erl b/src/rabbit_auth_backend_dummy.erl
index 5daca368..d2f07c1d 100644
--- a/src/rabbit_auth_backend_dummy.erl
+++ b/src/rabbit_auth_backend_dummy.erl
@@ -17,11 +17,12 @@
 -module(rabbit_auth_backend_dummy).
 -include("rabbit.hrl").
 
--behaviour(rabbit_auth_backend).
+-behaviour(rabbit_authn_backend).
+-behaviour(rabbit_authz_backend).
 
--export([description/0]).
 -export([user/0]).
--export([check_user_login/2, check_vhost_access/2, check_resource_access/3]).
+-export([user_login_authentication/2, user_login_authorization/1,
+         check_vhost_access/3, check_resource_access/3]).
 
 -ifdef(use_specs).
 
@@ -31,19 +32,17 @@
 
 %% A user to be used by the direct client when permission checks are
 %% not needed. This user can do anything AMQPish.
-user() -> #user{username     = <<"none">>,
-                tags         = [],
-                auth_backend = ?MODULE,
-                impl         = none}.
+user() -> #user{username       = <<"none">>,
+                tags           = [],
+                authz_backends = [{?MODULE, none}]}.
 
 %% Implementation of rabbit_auth_backend
 
-description() ->
-    [{name, <<"Dummy">>},
-     {description, <<"Database for the dummy user">>}].
+user_login_authentication(_, _) ->
+    {refused, "cannot log in conventionally as dummy user", []}.
 
-check_user_login(_, _) ->
+user_login_authorization(_) ->
     {refused, "cannot log in conventionally as dummy user", []}.
 
-check_vhost_access(#user{}, _VHostPath) -> true.
-check_resource_access(#user{}, #resource{}, _Permission) -> true.
+check_vhost_access(#auth_user{}, _VHostPath, _Sock) -> true.
+check_resource_access(#auth_user{}, #resource{}, _Permission) -> true.
diff --git a/src/rabbit_auth_backend_internal.erl b/src/rabbit_auth_backend_internal.erl
index fd1c4e8e..20a5766d 100644
--- a/src/rabbit_auth_backend_internal.erl
+++ b/src/rabbit_auth_backend_internal.erl
@@ -17,10 +17,11 @@
 -module(rabbit_auth_backend_internal).
 -include("rabbit.hrl").
 
--behaviour(rabbit_auth_backend).
+-behaviour(rabbit_authn_backend).
+-behaviour(rabbit_authz_backend).
 
--export([description/0]).
--export([check_user_login/2, check_vhost_access/2, check_resource_access/3]).
+-export([user_login_authentication/2, user_login_authorization/1,
+         check_vhost_access/3, check_resource_access/3]).
 
 -export([add_user/2, delete_user/1, lookup_user/1,
          change_password/2, clear_password/1,
@@ -76,13 +77,9 @@
 %%----------------------------------------------------------------------------
 %% Implementation of rabbit_auth_backend
 
-description() ->
-    [{name, <<"Internal">>},
-     {description, <<"Internal user / password database">>}].
-
-check_user_login(Username, []) ->
+user_login_authentication(Username, []) ->
     internal_check_user_login(Username, fun(_) -> true end);
-check_user_login(Username, [{password, Cleartext}]) ->
+user_login_authentication(Username, [{password, Cleartext}]) ->
     internal_check_user_login(
       Username,
       fun (#internal_user{password_hash = <<Salt:4/binary, Hash/binary>>}) ->
@@ -90,25 +87,30 @@ check_user_login(Username, [{password, Cleartext}]) ->
           (#internal_user{}) ->
               false
       end);
-check_user_login(Username, AuthProps) ->
+user_login_authentication(Username, AuthProps) ->
     exit({unknown_auth_props, Username, AuthProps}).
 
+user_login_authorization(Username) ->
+    case user_login_authentication(Username, []) of
+        {ok, #auth_user{impl = Impl}} -> {ok, Impl};
+        Else                          -> Else
+    end.
+
 internal_check_user_login(Username, Fun) ->
     Refused = {refused, "user '~s' - invalid credentials", [Username]},
     case lookup_user(Username) of
         {ok, User = #internal_user{tags = Tags}} ->
             case Fun(User) of
-                true -> {ok, #user{username     = Username,
-                                   tags         = Tags,
-                                   auth_backend = ?MODULE,
-                                   impl         = User}};
+                true -> {ok, #auth_user{username = Username,
+                                        tags     = Tags,
+                                        impl     = none}};
                 _    -> Refused
             end;
         {error, not_found} ->
             Refused
     end.
 
-check_vhost_access(#user{username = Username}, VHostPath) ->
+check_vhost_access(#auth_user{username = Username}, VHostPath, _Sock) ->
     case mnesia:dirty_read({rabbit_user_permission,
                             #user_vhost{username     = Username,
                                         virtual_host = VHostPath}}) of
@@ -116,7 +118,7 @@ check_vhost_access(#user{username = Username}, VHostPath) ->
         [_R] -> true
     end.
 
-check_resource_access(#user{username = Username},
+check_resource_access(#auth_user{username = Username},
                       #resource{virtual_host = VHostPath, name = Name},
                       Permission) ->
     case mnesia:dirty_read({rabbit_user_permission,
diff --git a/src/rabbit_auth_mechanism.erl b/src/rabbit_auth_mechanism.erl
index d11af095..c8e23a75 100644
--- a/src/rabbit_auth_mechanism.erl
+++ b/src/rabbit_auth_mechanism.erl
@@ -36,13 +36,13 @@
 %%     Another round is needed. Here's the state I want next time.
 %% {protocol_error, Msg, Args}
 %%     Client got the protocol wrong. Log and die.
-%% {refused, Msg, Args}
+%% {refused, Username, Msg, Args}
 %%     Client failed authentication. Log and die.
 -callback handle_response(binary(), any()) ->
     {'ok', rabbit_types:user()} |
     {'challenge', binary(), any()} |
     {'protocol_error', string(), [any()]} |
-    {'refused', string(), [any()]}.
+    {'refused', rabbit_types:username() | none, string(), [any()]}.
 
 -else.
 
diff --git a/src/rabbit_authn_backend.erl b/src/rabbit_authn_backend.erl
new file mode 100644
index 00000000..cfc3f5db
--- /dev/null
+++ b/src/rabbit_authn_backend.erl
@@ -0,0 +1,49 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2014 GoPivotal, Inc.  All rights reserved.
+%%
+
+-module(rabbit_authn_backend).
+
+-include("rabbit.hrl").
+
+-ifdef(use_specs).
+
+%% Check a user can log in, given a username and a proplist of
+%% authentication information (e.g. [{password, Password}]). If your
+%% backend is not to be used for authentication, this should always
+%% refuse access.
+%%
+%% Possible responses:
+%% {ok, User}
+%%     Authentication succeeded, and here's the user record.
+%% {error, Error}
+%%     Something went wrong. Log and die.
+%% {refused, Msg, Args}
+%%     Client failed authentication. Log and die.
+-callback user_login_authentication(rabbit_types:username(), [term()]) ->
+    {'ok', rabbit_types:auth_user()} |
+    {'refused', string(), [any()]} |
+    {'error', any()}.
+
+-else.
+
+-export([behaviour_info/1]).
+
+behaviour_info(callbacks) ->
+    [{user_login_authentication, 2}];
+behaviour_info(_Other) ->
+    undefined.
+
+-endif.
diff --git a/src/rabbit_auth_backend.erl b/src/rabbit_authz_backend.erl
index a7dd6494..ff5f014e 100644
--- a/src/rabbit_auth_backend.erl
+++ b/src/rabbit_authz_backend.erl
@@ -14,47 +14,49 @@
 %% Copyright (c) 2007-2014 GoPivotal, Inc.  All rights reserved.
 %%
 
--module(rabbit_auth_backend).
+-module(rabbit_authz_backend).
 
--ifdef(use_specs).
+-include("rabbit.hrl").
 
-%% A description proplist as with auth mechanisms,
-%% exchanges. Currently unused.
--callback description() -> [proplists:property()].
+-ifdef(use_specs).
 
-%% Check a user can log in, given a username and a proplist of
-%% authentication information (e.g. [{password, Password}]).
+%% Check a user can log in, when this backend is being used for
+%% authorisation only. Authentication has already taken place
+%% successfully, but we need to check that the user exists in this
+%% backend, and initialise any impl field we will want to have passed
+%% back in future calls to check_vhost_access/3 and
+%% check_resource_access/3.
 %%
 %% Possible responses:
-%% {ok, User}
-%%     Authentication succeeded, and here's the user record.
+%% {ok, Impl}
+%%     User authorisation succeeded, and here's the impl field.
 %% {error, Error}
 %%     Something went wrong. Log and die.
 %% {refused, Msg, Args}
-%%     Client failed authentication. Log and die.
--callback check_user_login(rabbit_types:username(), [term()]) ->
-    {'ok', rabbit_types:user()} |
+%%     User authorisation failed. Log and die.
+-callback user_login_authorization(rabbit_types:username()) ->
+    {'ok', any()} |
     {'refused', string(), [any()]} |
     {'error', any()}.
 
-%% Given #user and vhost, can a user log in to a vhost?
+%% Given #auth_user and vhost, can a user log in to a vhost?
 %% Possible responses:
 %% true
 %% false
 %% {error, Error}
 %%     Something went wrong. Log and die.
--callback check_vhost_access(rabbit_types:user(), rabbit_types:vhost()) ->
+-callback check_vhost_access(rabbit_types:auth_user(),
+                             rabbit_types:vhost(), rabbit_net:socket()) ->
     boolean() | {'error', any()}.
 
-
-%% Given #user, resource and permission, can a user access a resource?
+%% Given #auth_user, resource and permission, can a user access a resource?
 %%
 %% Possible responses:
 %% true
 %% false
 %% {error, Error}
 %%     Something went wrong. Log and die.
--callback check_resource_access(rabbit_types:user(),
+-callback check_resource_access(rabbit_types:auth_user(),
                                 rabbit_types:r(atom()),
                                 rabbit_access_control:permission_atom()) ->
     boolean() | {'error', any()}.
@@ -64,8 +66,8 @@
 -export([behaviour_info/1]).
 
 behaviour_info(callbacks) ->
-    [{description, 0}, {check_user_login, 2}, {check_vhost_access, 2},
-     {check_resource_access, 3}];
+    [{user_login_authorization, 1},
+     {check_vhost_access, 3}, {check_resource_access, 3}];
 behaviour_info(_Other) ->
     undefined.
 
diff --git a/src/rabbit_autoheal.erl b/src/rabbit_autoheal.erl
index 90458741..09e9aa6a 100644
--- a/src/rabbit_autoheal.erl
+++ b/src/rabbit_autoheal.erl
@@ -21,6 +21,8 @@
 %% The named process we are running in.
 -define(SERVER, rabbit_node_monitor).
 
+-define(MNESIA_STOPPED_PING_INTERNAL, 200).
+
 %%----------------------------------------------------------------------------
 
 %% In order to autoheal we want to:
@@ -54,9 +56,17 @@
 %%   - we are the winner and are waiting for all losing nodes to stop
 %%   before telling them they can restart
 %%
+%% about_to_heal
+%%   - we are the leader, and have already assigned the winner and
+%%   losers. We are part of the losers and we wait for the winner_is
+%%   announcement. This leader-specific state differs from not_healing
+%%   (the state other losers are in), because the leader could still
+%%   receive request_start messages: those subsequent requests must be
+%%   ignored.
+%%
 %% {leader_waiting, OutstandingStops}
 %%   - we are the leader, and have already assigned the winner and losers.
-%%     We are neither but need to ignore further requests to autoheal.
+%%   We are neither but need to ignore further requests to autoheal.
 %%
 %% restarting
 %%   - we are restarting. Of course the node monitor immediately dies
@@ -128,14 +138,12 @@ handle_msg({request_start, Node},
                             "  * Winner:     ~p~n"
                             "  * Losers:     ~p~n",
                             [AllPartitions, Winner, Losers]),
-            Continue = fun(Msg) ->
-                               handle_msg(Msg, not_healing, Partitions)
-                       end,
             case node() =:= Winner of
-                true  -> Continue({become_winner, Losers});
+                true  -> handle_msg({become_winner, Losers},
+                                    not_healing, Partitions);
                 false -> send(Winner, {become_winner, Losers}), %% [0]
                          case lists:member(node(), Losers) of
-                             true  -> Continue({winner_is, Winner});
+                             true  -> about_to_heal;
                              false -> {leader_waiting, Losers}
                          end
             end
@@ -163,7 +171,8 @@ handle_msg({become_winner, Losers},
     end;
 
 handle_msg({winner_is, Winner},
-           not_healing, _Partitions) ->
+           State, _Partitions)
+           when State =:= not_healing orelse State =:= about_to_heal ->
     rabbit_log:warning(
       "Autoheal: we were selected to restart; winner is ~p~n", [Winner]),
     rabbit_node_monitor:run_outside_applications(
@@ -194,9 +203,36 @@ abort(Down, Notify) ->
     winner_finish(Notify).
 
 winner_finish(Notify) ->
+    %% There is a race in Mnesia causing a starting loser to hang
+    %% forever if another loser stops at the same time: the starting
+    %% node connects to the other node, negotiates the protocol and
+    %% attempts to acquire a write lock on the schema on the other node.
+    %% If the other node stops between the protocol negotiation and lock
+    %% request, the starting node never gets an answer to its lock
+    %% request.
+    %%
+    %% To work around the problem, we make sure Mnesia is stopped on all
+    %% losing nodes before sending the "autoheal_safe_to_start" signal.
+    wait_for_mnesia_shutdown(Notify),
     [{rabbit_outside_app_process, N} ! autoheal_safe_to_start || N <- Notify],
     not_healing.
 
+wait_for_mnesia_shutdown([Node | Rest] = AllNodes) ->
+    case rpc:call(Node, mnesia, system_info, [is_running]) of
+        no ->
+            wait_for_mnesia_shutdown(Rest);
+        Running when
+        Running =:= yes orelse
+        Running =:= starting orelse
+        Running =:= stopping ->
+            timer:sleep(?MNESIA_STOPPED_PING_INTERNAL),
+            wait_for_mnesia_shutdown(AllNodes);
+        _ ->
+            wait_for_mnesia_shutdown(Rest)
+    end;
+wait_for_mnesia_shutdown([]) ->
+    ok.
+
 make_decision(AllPartitions) ->
     Sorted = lists:sort([{partition_value(P), P} || P <- AllPartitions]),
     [[Winner | _] | Rest] = lists:reverse([P || {_, P} <- Sorted]),
diff --git a/src/rabbit_binary_parser.erl b/src/rabbit_binary_parser.erl
index 3ab82cad..ee8147f4 100644
--- a/src/rabbit_binary_parser.erl
+++ b/src/rabbit_binary_parser.erl
@@ -41,48 +41,90 @@
 %% parse_table supports the AMQP 0-8/0-9 standard types, S, I, D, T
 %% and F, as well as the QPid extensions b, d, f, l, s, t, x, and V.
 
+-define(SIMPLE_PARSE_TABLE(BType, Pattern, RType),
+        parse_table(<<NLen:8/unsigned, NameString:NLen/binary,
+                      BType, Pattern, Rest/binary>>) ->
+               [{NameString, RType, Value} | parse_table(Rest)]).
+
+%% Note that we try to put these in approximately the order we expect
+%% to hit them, that's why the empty binary is half way through.
+
+parse_table(<<NLen:8/unsigned, NameString:NLen/binary,
+              $S, VLen:32/unsigned, Value:VLen/binary, Rest/binary>>) ->
+    [{NameString, longstr, Value} | parse_table(Rest)];
+
+?SIMPLE_PARSE_TABLE($I, Value:32/signed,   signedint);
+?SIMPLE_PARSE_TABLE($T, Value:64/unsigned, timestamp);
+
 parse_table(<<>>) ->
     [];
-parse_table(<<NLen:8/unsigned, NameString:NLen/binary, ValueAndRest/binary>>) ->
-    {Type, Value, Rest} = parse_field_value(ValueAndRest),
-    [{NameString, Type, Value} | parse_table(Rest)].
 
-parse_array(<<>>) ->
-    [];
-parse_array(<<ValueAndRest/binary>>) ->
-    {Type, Value, Rest} = parse_field_value(ValueAndRest),
-    [{Type, Value} | parse_array(Rest)].
+?SIMPLE_PARSE_TABLE($b, Value:8/signed,  byte);
+?SIMPLE_PARSE_TABLE($d, Value:64/float, double);
+?SIMPLE_PARSE_TABLE($f, Value:32/float, float);
+?SIMPLE_PARSE_TABLE($l, Value:64/signed, long);
+?SIMPLE_PARSE_TABLE($s, Value:16/signed, short);
+
+parse_table(<<NLen:8/unsigned, NameString:NLen/binary,
+              $t, Value:8/unsigned, Rest/binary>>) ->
+    [{NameString, bool, (Value /= 0)} | parse_table(Rest)];
+
+parse_table(<<NLen:8/unsigned, NameString:NLen/binary,
+              $D, Before:8/unsigned, After:32/unsigned, Rest/binary>>) ->
+    [{NameString, decimal, {Before, After}} | parse_table(Rest)];
 
-parse_field_value(<<$S, VLen:32/unsigned, V:VLen/binary, R/binary>>) ->
-    {longstr, V, R};
+parse_table(<<NLen:8/unsigned, NameString:NLen/binary,
+              $F, VLen:32/unsigned, Value:VLen/binary, Rest/binary>>) ->
+    [{NameString, table, parse_table(Value)} | parse_table(Rest)];
 
-parse_field_value(<<$I, V:32/signed, R/binary>>) ->
-    {signedint, V, R};
+parse_table(<<NLen:8/unsigned, NameString:NLen/binary,
+              $A, VLen:32/unsigned, Value:VLen/binary, Rest/binary>>) ->
+    [{NameString, array, parse_array(Value)} | parse_table(Rest)];
+
+parse_table(<<NLen:8/unsigned, NameString:NLen/binary,
+              $x, VLen:32/unsigned, Value:VLen/binary, Rest/binary>>) ->
+    [{NameString, binary, Value} | parse_table(Rest)];
+
+parse_table(<<NLen:8/unsigned, NameString:NLen/binary,
+              $V, Rest/binary>>) ->
+    [{NameString, void, undefined} | parse_table(Rest)].
+
+-define(SIMPLE_PARSE_ARRAY(BType, Pattern, RType),
+        parse_array(<<BType, Pattern, Rest/binary>>) ->
+               [{RType, Value} | parse_array(Rest)]).
+
+parse_array(<<$S, VLen:32/unsigned, Value:VLen/binary, Rest/binary>>) ->
+    [{longstr, Value} | parse_array(Rest)];
+
+?SIMPLE_PARSE_ARRAY($I, Value:32/signed,   signedint);
+?SIMPLE_PARSE_ARRAY($T, Value:64/unsigned, timestamp);
+
+parse_array(<<>>) ->
+    [];
 
-parse_field_value(<<$D, Before:8/unsigned, After:32/unsigned, R/binary>>) ->
-    {decimal, {Before, After}, R};
+?SIMPLE_PARSE_ARRAY($b, Value:8/signed,  byte);
+?SIMPLE_PARSE_ARRAY($d, Value:64/float, double);
+?SIMPLE_PARSE_ARRAY($f, Value:32/float, float);
+?SIMPLE_PARSE_ARRAY($l, Value:64/signed, long);
+?SIMPLE_PARSE_ARRAY($s, Value:16/signed, short);
 
-parse_field_value(<<$T, V:64/unsigned, R/binary>>) ->
-    {timestamp, V, R};
+parse_array(<<$t, Value:8/unsigned, Rest/binary>>) ->
+    [{bool, (Value /= 0)} | parse_array(Rest)];
 
-parse_field_value(<<$F, VLen:32/unsigned, Table:VLen/binary, R/binary>>) ->
-    {table, parse_table(Table), R};
+parse_array(<<$D, Before:8/unsigned, After:32/unsigned, Rest/binary>>) ->
+    [{decimal, {Before, After}} | parse_array(Rest)];
 
-parse_field_value(<<$A, VLen:32/unsigned, Array:VLen/binary, R/binary>>) ->
-    {array, parse_array(Array), R};
+parse_array(<<$F, VLen:32/unsigned, Value:VLen/binary, Rest/binary>>) ->
+    [{table, parse_table(Value)} | parse_array(Rest)];
 
-parse_field_value(<<$b, V:8/signed,   R/binary>>) -> {byte,        V, R};
-parse_field_value(<<$d, V:64/float,   R/binary>>) -> {double,      V, R};
-parse_field_value(<<$f, V:32/float,   R/binary>>) -> {float,       V, R};
-parse_field_value(<<$l, V:64/signed,  R/binary>>) -> {long,        V, R};
-parse_field_value(<<$s, V:16/signed,  R/binary>>) -> {short,       V, R};
-parse_field_value(<<$t, V:8/unsigned, R/binary>>) -> {bool, (V /= 0), R};
+parse_array(<<$A, VLen:32/unsigned, Value:VLen/binary, Rest/binary>>) ->
+    [{array, parse_array(Value)} | parse_array(Rest)];
 
-parse_field_value(<<$x, VLen:32/unsigned, V:VLen/binary, R/binary>>) ->
-    {binary, V, R};
+parse_array(<<$x, VLen:32/unsigned, Value:VLen/binary, Rest/binary>>) ->
+    [{binary, Value} | parse_array(Rest)];
 
-parse_field_value(<<$V, R/binary>>) ->
-    {void, undefined, R}.
+parse_array(<<$V, Rest/binary>>) ->
+    [{void, undefined} | parse_array(Rest)].
 
 ensure_content_decoded(Content = #content{properties = Props})
   when Props =/= none ->
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 8632e1b3..13cc925c 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -581,7 +581,8 @@ check_user_id_header(#'P_basic'{user_id = Username},
                      #ch{user = #user{username = Username}}) ->
     ok;
 check_user_id_header(
-  #'P_basic'{}, #ch{user = #user{auth_backend = rabbit_auth_backend_dummy}}) ->
+  #'P_basic'{}, #ch{user = #user{authz_backends =
+                                     [{rabbit_auth_backend_dummy, _}]}}) ->
     ok;
 check_user_id_header(#'P_basic'{user_id = Claimed},
                      #ch{user = #user{username = Actual,
diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
index a931eef0..bca740c6 100644
--- a/src/rabbit_control_main.erl
+++ b/src/rabbit_control_main.erl
@@ -19,7 +19,7 @@
 -include("rabbit_cli.hrl").
 
 -export([start/0, stop/0, parse_arguments/2, action/5,
-         sync_queue/1, cancel_sync_queue/1]).
+         sync_queue/1, cancel_sync_queue/1, become/1]).
 
 -import(rabbit_cli, [rpc_call/4]).
 
@@ -40,6 +40,7 @@
          change_cluster_node_type,
          update_cluster_nodes,
          {forget_cluster_node, [?OFFLINE_DEF]},
+         rename_cluster_node,
          force_boot,
          cluster_status,
          {sync_queue, [?VHOST_DEF]},
@@ -104,8 +105,8 @@
 -define(COMMANDS_NOT_REQUIRING_APP,
         [stop, stop_app, start_app, wait, reset, force_reset, rotate_logs,
          join_cluster, change_cluster_node_type, update_cluster_nodes,
-         forget_cluster_node, cluster_status, status, environment, eval,
-         force_boot]).
+         forget_cluster_node, rename_cluster_node, cluster_status, status,
+         environment, eval, force_boot]).
 
 %%----------------------------------------------------------------------------
 
@@ -234,6 +235,13 @@ action(forget_cluster_node, Node, [ClusterNodeS], Opts, Inform) ->
                           [ClusterNode, false])
     end;
 
+action(rename_cluster_node, Node, NodesS, _Opts, Inform) ->
+    Nodes = split_list([list_to_atom(N) || N <- NodesS]),
+    Inform("Renaming cluster nodes:~n~s~n",
+           [lists:flatten([rabbit_misc:format("  ~s -> ~s~n", [F, T]) ||
+                              {F, T} <- Nodes])]),
+    rabbit_mnesia_rename:rename(Node, Nodes);
+
 action(force_boot, Node, [], _Opts, Inform) ->
     Inform("Forcing boot for Mnesia dir ~s", [mnesia:system_info(directory)]),
     case rabbit:is_running(Node) of
@@ -591,11 +599,11 @@ start_distribution() ->
     {ok, _} = net_kernel:start([list_to_atom(CtlNodeName), name_type()]).
 
 become(BecomeNode) ->
+    error_logger:tty(false),
+    ok = net_kernel:stop(),
     case net_adm:ping(BecomeNode) of
         pong -> exit({node_running, BecomeNode});
         pang -> io:format("  * Impersonating node: ~s...", [BecomeNode]),
-                error_logger:tty(false),
-                ok = net_kernel:stop(),
                 {ok, _} = net_kernel:start([BecomeNode, name_type()]),
                 io:format(" done~n", []),
                 Dir = mnesia:system_info(directory),
@@ -720,3 +728,7 @@ prettify_typed_amqp_value(table,   Value) -> prettify_amqp_table(Value);
 prettify_typed_amqp_value(array,   Value) -> [prettify_typed_amqp_value(T, V) ||
                                                  {T, V} <- Value];
 prettify_typed_amqp_value(_Type,   Value) -> Value.
+
+split_list([])         -> [];
+split_list([_])        -> exit(even_list_needed);
+split_list([A, B | T]) -> [{A, B} | split_list(T)].
diff --git a/src/rabbit_diagnostics.erl b/src/rabbit_diagnostics.erl
index bf45b757..9fc0fabd 100644
--- a/src/rabbit_diagnostics.erl
+++ b/src/rabbit_diagnostics.erl
@@ -17,10 +17,11 @@
 -module(rabbit_diagnostics).
 
 -define(PROCESS_INFO,
-        [current_stacktrace, initial_call, dictionary, message_queue_len,
-         links, monitors, monitored_by, heap_size]).
+        [registered_name, current_stacktrace, initial_call, dictionary,
+         message_queue_len, links, monitors, monitored_by, heap_size]).
 
--export([maybe_stuck/0, maybe_stuck/1]).
+-export([maybe_stuck/0, maybe_stuck/1, top_memory_use/0, top_memory_use/1,
+         top_binary_refs/0, top_binary_refs/1]).
 
 maybe_stuck() -> maybe_stuck(5000).
 
@@ -41,13 +42,13 @@ maybe_stuck(Pids, Timeout) ->
     maybe_stuck(Pids2, Timeout - 500).
 
 looks_stuck(Pid) ->
-    case catch process_info(Pid, status) of
+    case info(Pid, status, gone) of
         {status, waiting} ->
             %% It's tempting to just check for message_queue_len > 0
             %% here rather than mess around with stack traces and
             %% heuristics. But really, sometimes freshly stuck
             %% processes can have 0 messages...
-            case catch erlang:process_info(Pid, current_stacktrace) of
+            case info(Pid, current_stacktrace, gone) of
                 {current_stacktrace, [H|_]} ->
                     maybe_stuck_stacktrace(H);
                 _ ->
@@ -75,5 +76,38 @@ maybe_stuck_stacktrace({_M, F, _A}) ->
         _ -> false
     end.
 
+top_memory_use() -> top_memory_use(30).
+
+top_memory_use(Count) ->
+    Pids = processes(),
+    io:format("Memory use: top ~p of ~p processes.~n", [Count, length(Pids)]),
+    Procs = [{info(Pid, memory, 0), info(Pid)} || Pid <- Pids],
+    Sorted = lists:sublist(lists:reverse(lists:sort(Procs)), Count),
+    io:format("~p~n", [Sorted]).
+
+top_binary_refs() -> top_binary_refs(30).
+
+top_binary_refs(Count) ->
+    Pids = processes(),
+    io:format("Binary refs: top ~p of ~p processes.~n", [Count, length(Pids)]),
+    Procs = [{{binary_refs, binary_refs(Pid)}, info(Pid)} || Pid <- Pids],
+    Sorted = lists:sublist(lists:reverse(lists:sort(Procs)), Count),
+    io:format("~p~n", [Sorted]).
+
+binary_refs(Pid) ->
+    {binary, Refs} = info(Pid, binary, []),
+    lists:sum([Sz || {_Ptr, Sz} <- lists:usort([{Ptr, Sz} ||
+                                                   {Ptr, Sz, _Cnt} <- Refs])]).
+
 info(Pid) ->
-    [{pid, Pid} | process_info(Pid, ?PROCESS_INFO)].
+    [{pid, Pid} | info(Pid, ?PROCESS_INFO, [])].
+
+info(Pid, Infos, Default) ->
+    try
+        process_info(Pid, Infos)
+    catch
+        _:_ -> case is_atom(Infos) of
+                   true  -> {Infos, Default};
+                   false -> Default
+               end
+    end.
diff --git a/src/rabbit_direct.erl b/src/rabbit_direct.erl
index 749a67b1..11233e7e 100644
--- a/src/rabbit_direct.erl
+++ b/src/rabbit_direct.erl
@@ -83,16 +83,27 @@ connect({Username, Password}, VHost, Protocol, Pid, Infos) ->
 connect0(AuthFun, VHost, Protocol, Pid, Infos) ->
     case rabbit:is_running() of
         true  -> case AuthFun() of
-                     {ok, User} ->
+                     {ok, User = #user{username = Username}} ->
+                         notify_auth_result(Username,
+                           user_authentication_success, []),
                          connect1(User, VHost, Protocol, Pid, Infos);
-                     {refused, _M, _A} ->
+                     {refused, Username, Msg, Args} ->
+                         notify_auth_result(Username,
+                           user_authentication_failure,
+                           [{error, rabbit_misc:format(Msg, Args)}]),
                          {error, {auth_failure, "Refused"}}
                  end;
         false -> {error, broker_not_found_on_node}
     end.
 
+notify_auth_result(Username, AuthResult, ExtraProps) ->
+    EventProps = [{connection_type, direct},
+                  {name, case Username of none -> ''; _ -> Username end}] ++
+                 ExtraProps,
+    rabbit_event:notify(AuthResult, [P || {_, V} = P <- EventProps, V =/= '']).
+
 connect1(User, VHost, Protocol, Pid, Infos) ->
-    try rabbit_access_control:check_vhost_access(User, VHost) of
+    try rabbit_access_control:check_vhost_access(User, VHost, undefined) of
         ok -> ok = pg_local:join(rabbit_direct, Pid),
               rabbit_event:notify(connection_created, Infos),
               {ok, {User, rabbit_reader:server_properties(Protocol)}}
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 3e2c88ee..20d7051c 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -44,7 +44,8 @@
 -export([format/2, format_many/1, format_stderr/2]).
 -export([unfold/2, ceil/1, queue_fold/3]).
 -export([sort_field_table/1]).
--export([pid_to_string/1, string_to_pid/1, node_to_fake_pid/1]).
+-export([pid_to_string/1, string_to_pid/1,
+         pid_change_node/2, node_to_fake_pid/1]).
 -export([version_compare/2, version_compare/3]).
 -export([version_minor_equivalent/2]).
 -export([dict_cons/3, orddict_cons/3, gb_trees_cons/3]).
@@ -196,6 +197,7 @@
         (rabbit_framing:amqp_table()) -> rabbit_framing:amqp_table()).
 -spec(pid_to_string/1 :: (pid()) -> string()).
 -spec(string_to_pid/1 :: (string()) -> pid()).
+-spec(pid_change_node/2 :: (pid(), node()) -> pid()).
 -spec(node_to_fake_pid/1 :: (atom()) -> pid()).
 -spec(version_compare/2 :: (string(), string()) -> 'lt' | 'eq' | 'gt').
 -spec(version_compare/3 ::
@@ -686,11 +688,7 @@ sort_field_table(Arguments) ->
 %% regardless of what node we are running on. The representation also
 %% permits easy identification of the pid's node.
 pid_to_string(Pid) when is_pid(Pid) ->
-    %% see http://erlang.org/doc/apps/erts/erl_ext_dist.html (8.10 and
-    %% 8.7)
-    <<131,103,100,NodeLen:16,NodeBin:NodeLen/binary,Id:32,Ser:32,Cre:8>>
-        = term_to_binary(Pid),
-    Node = binary_to_term(<<131,100,NodeLen:16,NodeBin:NodeLen/binary>>),
+    {Node, Cre, Id, Ser} = decompose_pid(Pid),
     format("<~s.~B.~B.~B>", [Node, Cre, Id, Ser]).
 
 %% inverse of above
@@ -701,17 +699,32 @@ string_to_pid(Str) ->
     case re:run(Str, "^<(.*)\\.(\\d+)\\.(\\d+)\\.(\\d+)>\$",
                 [{capture,all_but_first,list}]) of
         {match, [NodeStr, CreStr, IdStr, SerStr]} ->
-            <<131,NodeEnc/binary>> = term_to_binary(list_to_atom(NodeStr)),
             [Cre, Id, Ser] = lists:map(fun list_to_integer/1,
                                        [CreStr, IdStr, SerStr]),
-            binary_to_term(<<131,103,NodeEnc/binary,Id:32,Ser:32,Cre:8>>);
+            compose_pid(list_to_atom(NodeStr), Cre, Id, Ser);
         nomatch ->
             throw(Err)
     end.
 
+pid_change_node(Pid, NewNode) ->
+    {_OldNode, Cre, Id, Ser} = decompose_pid(Pid),
+    compose_pid(NewNode, Cre, Id, Ser).
+
 %% node(node_to_fake_pid(Node)) =:= Node.
 node_to_fake_pid(Node) ->
-    string_to_pid(format("<~s.0.0.0>", [Node])).
+    compose_pid(Node, 0, 0, 0).
+
+decompose_pid(Pid) when is_pid(Pid) ->
+    %% see http://erlang.org/doc/apps/erts/erl_ext_dist.html (8.10 and
+    %% 8.7)
+    <<131,103,100,NodeLen:16,NodeBin:NodeLen/binary,Id:32,Ser:32,Cre:8>>
+        = term_to_binary(Pid),
+    Node = binary_to_term(<<131,100,NodeLen:16,NodeBin:NodeLen/binary>>),
+    {Node, Cre, Id, Ser}.
+
+compose_pid(Node, Cre, Id, Ser) ->
+    <<131,NodeEnc/binary>> = term_to_binary(Node),
+    binary_to_term(<<131,103,NodeEnc/binary,Id:32,Ser:32,Cre:8>>).
 
 version_compare(A, B, lte) ->
     case version_compare(A, B) of
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index fa51dd70..f9110e58 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -113,23 +113,29 @@ init() ->
     ok.
 
 init_from_config() ->
+    FindBadNodeNames = fun
+        (Name, BadNames) when is_atom(Name) -> BadNames;
+        (Name, BadNames)                    -> [Name | BadNames]
+    end,
     {TryNodes, NodeType} =
         case application:get_env(rabbit, cluster_nodes) of
+            {ok, {Nodes, Type} = Config}
+            when is_list(Nodes) andalso (Type == disc orelse Type == ram) ->
+                case lists:foldr(FindBadNodeNames, [], Nodes) of
+                    []       -> Config;
+                    BadNames -> e({invalid_cluster_node_names, BadNames})
+                end;
+            {ok, {_, BadType}} when BadType /= disc andalso BadType /= ram ->
+                e({invalid_cluster_node_type, BadType});
             {ok, Nodes} when is_list(Nodes) ->
-                Config = {Nodes -- [node()], case lists:member(node(), Nodes) of
-                                                 true  -> disc;
-                                                 false -> ram
-                                             end},
-                rabbit_log:warning(
-                  "Converting legacy 'cluster_nodes' configuration~n    ~w~n"
-                  "to~n    ~w.~n~n"
-                  "Please update the configuration to the new format "
-                  "{Nodes, NodeType}, where Nodes contains the nodes that the "
-                  "node will try to cluster with, and NodeType is either "
-                  "'disc' or 'ram'~n", [Nodes, Config]),
-                Config;
-            {ok, Config} ->
-                Config
+                %% The legacy syntax (a nodes list without the node
+                %% type) is unsupported.
+                case lists:foldr(FindBadNodeNames, [], Nodes) of
+                    [] -> e(cluster_node_type_mandatory);
+                    _  -> e(invalid_cluster_nodes_conf)
+                end;
+            {ok, _} ->
+                e(invalid_cluster_nodes_conf)
         end,
     case TryNodes of
         [] -> init_db_and_upgrade([node()], disc, false);
@@ -174,18 +180,25 @@ join_cluster(DiscoveryNode, NodeType) ->
     {ClusterNodes, _, _} = discover_cluster([DiscoveryNode]),
     case me_in_nodes(ClusterNodes) of
         false ->
-            %% reset the node. this simplifies things and it will be needed in
-            %% this case - we're joining a new cluster with new nodes which
-            %% are not in synch with the current node. I also lifts the burden
-            %% of reseting the node from the user.
-            reset_gracefully(),
-
-            %% Join the cluster
-            rabbit_log:info("Clustering with ~p as ~p node~n",
-                            [ClusterNodes, NodeType]),
-            ok = init_db_with_mnesia(ClusterNodes, NodeType, true, true),
-            rabbit_node_monitor:notify_joined_cluster(),
-            ok;
+            case check_cluster_consistency(DiscoveryNode, false) of
+                {ok, _} ->
+                    %% reset the node. this simplifies things and it
+                    %% will be needed in this case - we're joining a new
+                    %% cluster with new nodes which are not in synch
+                    %% with the current node. It also lifts the burden
+                    %% of resetting the node from the user.
+                    reset_gracefully(),
+
+                    %% Join the cluster
+                    rabbit_log:info("Clustering with ~p as ~p node~n",
+                                    [ClusterNodes, NodeType]),
+                    ok = init_db_with_mnesia(ClusterNodes, NodeType,
+                                             true, true),
+                    rabbit_node_monitor:notify_joined_cluster(),
+                    ok;
+                {error, Reason} ->
+                    {error, Reason}
+            end;
         true ->
             rabbit_log:info("Already member of cluster: ~p~n", [ClusterNodes]),
             {ok, already_member}
@@ -545,7 +558,7 @@ maybe_force_load() ->
 check_cluster_consistency() ->
     %% We want to find 0 or 1 consistent nodes.
     case lists:foldl(
-           fun (Node,  {error, _})    -> check_cluster_consistency(Node);
+           fun (Node,  {error, _})    -> check_cluster_consistency(Node, true);
                (_Node, {ok, Status})  -> {ok, Status}
            end, {error, not_found}, nodes_excl_me(cluster_nodes(all)))
     of
@@ -575,17 +588,22 @@ check_cluster_consistency() ->
             throw(E)
     end.
 
-check_cluster_consistency(Node) ->
+check_cluster_consistency(Node, CheckNodesConsistency) ->
     case rpc:call(Node, rabbit_mnesia, node_info, []) of
         {badrpc, _Reason} ->
             {error, not_found};
         {_OTP, _Rabbit, {error, _}} ->
             {error, not_found};
-        {OTP, Rabbit, {ok, Status}} ->
+        {OTP, Rabbit, {ok, Status}} when CheckNodesConsistency ->
             case check_consistency(OTP, Rabbit, Node, Status) of
                 {error, _} = E -> E;
                 {ok, Res}      -> {ok, Res}
             end;
+        {OTP, Rabbit, {ok, Status}} ->
+            case check_consistency(OTP, Rabbit) of
+                {error, _} = E -> E;
+                ok             -> {ok, Status}
+            end;
         {_OTP, Rabbit, _Hash, _Status} ->
             %% delegate hash checking implies version mismatch
             version_error("Rabbit", rabbit_misc:version(), Rabbit)
@@ -838,6 +856,20 @@ nodes_excl_me(Nodes) -> Nodes -- [node()].
 
 e(Tag) -> throw({error, {Tag, error_description(Tag)}}).
 
+error_description({invalid_cluster_node_names, BadNames}) ->
+    "In the 'cluster_nodes' configuration key, the following node names "
+        "are invalid: " ++ lists:flatten(io_lib:format("~p", [BadNames]));
+error_description({invalid_cluster_node_type, BadType}) ->
+    "In the 'cluster_nodes' configuration key, the node type is invalid "
+        "(expected 'disc' or 'ram'): " ++
+        lists:flatten(io_lib:format("~p", [BadType]));
+error_description(cluster_node_type_mandatory) ->
+    "The 'cluster_nodes' configuration key must indicate the node type: "
+        "either {[...], disc} or {[...], ram}";
+error_description(invalid_cluster_nodes_conf) ->
+    "The 'cluster_nodes' configuration key is invalid, it must be of the "
+        "form {[Nodes], Type}, where Nodes is a list of node names and "
+        "Type is either 'disc' or 'ram'";
 error_description(clustering_only_disc_node) ->
     "You cannot cluster a node if it is the only disc node in its existing "
         " cluster. If new nodes joined while this node was offline, use "
diff --git a/src/rabbit_mnesia_rename.erl b/src/rabbit_mnesia_rename.erl
new file mode 100644
index 00000000..2787cb74
--- /dev/null
+++ b/src/rabbit_mnesia_rename.erl
@@ -0,0 +1,267 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2014 GoPivotal, Inc.  All rights reserved.
+%%
+
+-module(rabbit_mnesia_rename).
+-include("rabbit.hrl").
+
+-export([rename/2]).
+-export([maybe_finish/1]).
+
+-define(CONVERT_TABLES, [schema, rabbit_durable_queue]).
+
+%% Supports renaming the nodes in the Mnesia database. In order to do
+%% this, we take a backup of the database, traverse the backup
+%% changing node names and pids as we go, then restore it.
+%%
+%% That's enough for a standalone node, for clusters the story is more
+%% complex. We can take pairs of nodes From and To, but backing up and
+%% restoring the database changes schema cookies, so if we just do
+%% this on all nodes the cluster will refuse to re-form with
+%% "Incompatible schema cookies.". Therefore we do something similar
+%% to what we do for upgrades - the first node in the cluster to
+%% restart becomes the authority, and other nodes wipe their own
+%% Mnesia state and rejoin. They also need to tell Mnesia the old node
+%% is not coming back.
+%%
+%% If we are renaming nodes one at a time then the running cluster
+%% might not be aware that a rename has taken place, so after we wipe
+%% and rejoin we then update any tables (in practice just
+%% rabbit_durable_queue) which should be aware that we have changed.
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(rename/2 :: (node(), [{node(), node()}]) -> 'ok').
+-spec(maybe_finish/1 :: ([node()]) -> 'ok').
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+rename(Node, NodeMapList) ->
+    try
+        %% Check everything is correct and figure out what we are
+        %% changing from and to.
+        {FromNode, ToNode, NodeMap} = prepare(Node, NodeMapList),
+
+        %% We backup and restore Mnesia even if other nodes are
+        %% running at the time, and defer the final decision about
+        %% whether to use our mutated copy or rejoin the cluster until
+        %% we restart. That means we might be mutating our copy of the
+        %% database while the cluster is running. *Do not* contact the
+        %% cluster while this is happening, we are likely to get
+        %% confused.
+        application:set_env(kernel, dist_auto_connect, never),
+
+        %% Take a copy we can restore from if we abandon the
+        %% rename. We don't restore from the "backup" since restoring
+        %% that changes schema cookies and might stop us rejoining the
+        %% cluster.
+        ok = rabbit_mnesia:copy_db(mnesia_copy_dir()),
+
+        %% And make the actual changes
+        rabbit_control_main:become(FromNode),
+        take_backup(before_backup_name()),
+        convert_backup(NodeMap, before_backup_name(), after_backup_name()),
+        ok = rabbit_file:write_term_file(rename_config_name(),
+                                         [{FromNode, ToNode}]),
+        convert_config_files(NodeMap),
+        rabbit_control_main:become(ToNode),
+        restore_backup(after_backup_name()),
+        ok
+    after
+        stop_mnesia()
+    end.
+
+prepare(Node, NodeMapList) ->
+    %% If we have a previous rename and haven't started since, give up.
+    case rabbit_file:is_dir(dir()) of
+        true  -> exit({rename_in_progress,
+                       "Restart node under old name to roll back"});
+        false -> ok = rabbit_file:ensure_dir(mnesia_copy_dir())
+    end,
+
+    %% Check we don't have two nodes mapped to the same node
+    {FromNodes, ToNodes} = lists:unzip(NodeMapList),
+    case length(FromNodes) - length(lists:usort(ToNodes)) of
+        0 -> ok;
+        _ -> exit({duplicate_node, ToNodes})
+    end,
+
+    %% Figure out which node we are before and after the change
+    FromNode = case [From || {From, To} <- NodeMapList,
+                             To =:= Node] of
+                   [N] -> N;
+                   []  -> Node
+               end,
+    NodeMap = dict:from_list(NodeMapList),
+    ToNode = case dict:find(FromNode, NodeMap) of
+                 {ok, N2} -> N2;
+                 error    -> FromNode
+             end,
+
+    %% Check that we are in the cluster, all old nodes are in the
+    %% cluster, and no new nodes are.
+    Nodes = rabbit_mnesia:cluster_nodes(all),
+    case {FromNodes -- Nodes, ToNodes -- (ToNodes -- Nodes),
+          lists:member(Node, Nodes ++ ToNodes)} of
+        {[], [], true}  -> ok;
+        {[], [], false} -> exit({i_am_not_involved,        Node});
+        {F,  [], _}     -> exit({nodes_not_in_cluster,     F});
+        {_,  T,  _}     -> exit({nodes_already_in_cluster, T})
+    end,
+    {FromNode, ToNode, NodeMap}.
+
+take_backup(Backup) ->
+    start_mnesia(),
+    ok = mnesia:backup(Backup),
+    stop_mnesia().
+
+restore_backup(Backup) ->
+    ok = mnesia:install_fallback(Backup, [{scope, local}]),
+    start_mnesia(),
+    stop_mnesia(),
+    rabbit_mnesia:force_load_next_boot().
+
+maybe_finish(AllNodes) ->
+    case rabbit_file:read_term_file(rename_config_name()) of
+        {ok, [{FromNode, ToNode}]} -> finish(FromNode, ToNode, AllNodes);
+        _                          -> ok
+    end.
+
+finish(FromNode, ToNode, AllNodes) ->
+    case node() of
+        ToNode ->
+            case rabbit_upgrade:nodes_running(AllNodes) of
+                [] -> finish_primary(FromNode, ToNode);
+                _  -> finish_secondary(FromNode, ToNode, AllNodes)
+            end;
+        FromNode ->
+            rabbit_log:info(
+              "Abandoning rename from ~s to ~s since we are still ~s~n",
+              [FromNode, ToNode, FromNode]),
+            [{ok, _} = file:copy(backup_of_conf(F), F) || F <- config_files()],
+            ok = rabbit_file:recursive_delete([rabbit_mnesia:dir()]),
+            ok = rabbit_file:recursive_copy(
+                   mnesia_copy_dir(), rabbit_mnesia:dir()),
+            delete_rename_files();
+        _ ->
+            %% Boot will almost certainly fail but we might as
+            %% well just log this
+            rabbit_log:info(
+              "Rename attempted from ~s to ~s but we are ~s - ignoring.~n",
+              [FromNode, ToNode, node()])
+    end.
+
+finish_primary(FromNode, ToNode) ->
+    rabbit_log:info("Restarting as primary after rename from ~s to ~s~n",
+                    [FromNode, ToNode]),
+    delete_rename_files(),
+    ok.
+
+finish_secondary(FromNode, ToNode, AllNodes) ->
+    rabbit_log:info("Restarting as secondary after rename from ~s to ~s~n",
+                    [FromNode, ToNode]),
+    rabbit_upgrade:secondary_upgrade(AllNodes),
+    rename_in_running_mnesia(FromNode, ToNode),
+    delete_rename_files(),
+    ok.
+
+dir()                -> rabbit_mnesia:dir() ++ "-rename".
+before_backup_name() -> dir() ++ "/backup-before".
+after_backup_name()  -> dir() ++ "/backup-after".
+rename_config_name() -> dir() ++ "/pending.config".
+mnesia_copy_dir()    -> dir() ++ "/mnesia-copy".
+
+delete_rename_files() -> ok = rabbit_file:recursive_delete([dir()]).
+
+start_mnesia() -> rabbit_misc:ensure_ok(mnesia:start(), cannot_start_mnesia),
+                  rabbit_table:force_load(),
+                  rabbit_table:wait_for_replicated().
+stop_mnesia()  -> stopped = mnesia:stop().
+
+convert_backup(NodeMap, FromBackup, ToBackup) ->
+    mnesia:traverse_backup(
+      FromBackup, ToBackup,
+      fun
+          (Row, Acc) ->
+              case lists:member(element(1, Row), ?CONVERT_TABLES) of
+                  true  -> {[update_term(NodeMap, Row)], Acc};
+                  false -> {[Row], Acc}
+              end
+      end, switched).
+
+config_files() ->
+    [rabbit_node_monitor:running_nodes_filename(),
+     rabbit_node_monitor:cluster_status_filename()].
+
+backup_of_conf(Path) ->
+    filename:join([dir(), filename:basename(Path)]).
+
+convert_config_files(NodeMap) ->
+    [convert_config_file(NodeMap, Path) || Path <- config_files()].
+
+convert_config_file(NodeMap, Path) ->
+    {ok, Term} = rabbit_file:read_term_file(Path),
+    {ok, _} = file:copy(Path, backup_of_conf(Path)),
+    ok = rabbit_file:write_term_file(Path, update_term(NodeMap, Term)).
+
+lookup_node(OldNode, NodeMap) ->
+    case dict:find(OldNode, NodeMap) of
+        {ok, NewNode} -> NewNode;
+        error         -> OldNode
+    end.
+
+mini_map(FromNode, ToNode) -> dict:from_list([{FromNode, ToNode}]).
+
+update_term(NodeMap, L) when is_list(L) ->
+    [update_term(NodeMap, I) || I <- L];
+update_term(NodeMap, T) when is_tuple(T) ->
+    list_to_tuple(update_term(NodeMap, tuple_to_list(T)));
+update_term(NodeMap, Node) when is_atom(Node) ->
+    lookup_node(Node, NodeMap);
+update_term(NodeMap, Pid) when is_pid(Pid) ->
+    rabbit_misc:pid_change_node(Pid, lookup_node(node(Pid), NodeMap));
+update_term(_NodeMap, Term) ->
+    Term.
+
+rename_in_running_mnesia(FromNode, ToNode) ->
+    All = rabbit_mnesia:cluster_nodes(all),
+    Running = rabbit_mnesia:cluster_nodes(running),
+    case {lists:member(FromNode, Running), lists:member(ToNode, All)} of
+        {false, true}  -> ok;
+        {true,  _}     -> exit({old_node_running,        FromNode});
+        {_,     false} -> exit({new_node_not_in_cluster, ToNode})
+    end,
+    {atomic, ok} = mnesia:del_table_copy(schema, FromNode),
+    Map = mini_map(FromNode, ToNode),
+    {atomic, _} = transform_table(rabbit_durable_queue, Map),
+    ok.
+
+transform_table(Table, Map) ->
+    mnesia:sync_transaction(
+      fun () ->
+              mnesia:lock({table, Table}, write),
+              transform_table(Table, Map, mnesia:first(Table))
+      end).
+
+transform_table(_Table, _Map, '$end_of_table') ->
+    ok;
+transform_table(Table, Map, Key) ->
+    [Term] = mnesia:read(Table, Key, write),
+    ok = mnesia:write(Table, update_term(Map, Term), write),
+    transform_table(Table, Map, mnesia:next(Table, Key)).
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index b829ae94..6c80ddcd 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -1299,7 +1299,8 @@ should_mask_action(CRef, MsgId,
 
 open_file(Dir, FileName, Mode) ->
     file_handle_cache:open(form_filename(Dir, FileName), ?BINARY_MODE ++ Mode,
-                           [{write_buffer, ?HANDLE_CACHE_BUFFER_SIZE}]).
+                           [{write_buffer, ?HANDLE_CACHE_BUFFER_SIZE},
+                            {read_buffer,  ?HANDLE_CACHE_BUFFER_SIZE}]).
 
 close_handle(Key, CState = #client_msstate { file_handle_cache = FHC }) ->
     CState #client_msstate { file_handle_cache = close_handle(Key, FHC) };
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index d59b22f6..1a288374 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -188,12 +188,21 @@ fix_ssl_options(Config) ->
     fix_verify_fun(fix_ssl_protocol_versions(Config)).
 
 fix_verify_fun(SslOptsConfig) ->
+    %% Starting with ssl 4.0.1 in Erlang R14B, the verify_fun function
+    %% takes 3 arguments and returns a tuple.
+    {ok, SslAppVer} = application:get_key(ssl, vsn),
+    UseNewVerifyFun = rabbit_misc:version_compare(SslAppVer, "4.0.1", gte),
     case rabbit_misc:pget(verify_fun, SslOptsConfig) of
+        {Module, Function, InitialUserState} ->
+            Fun = make_verify_fun(Module, Function, InitialUserState,
+                                  UseNewVerifyFun),
+            rabbit_misc:pset(verify_fun, Fun, SslOptsConfig);
         {Module, Function} ->
-            rabbit_misc:pset(verify_fun,
-                             fun (ErrorList) ->
-                                     Module:Function(ErrorList)
-                             end, SslOptsConfig);
+            Fun = make_verify_fun(Module, Function, none,
+                                  UseNewVerifyFun),
+            rabbit_misc:pset(verify_fun, Fun, SslOptsConfig);
+        undefined when UseNewVerifyFun ->
+            SslOptsConfig;
         undefined ->
             % unknown_ca errors are silently ignored prior to R14B unless we
             % supply this verify_fun - remove when at least R14B is required
@@ -206,6 +215,48 @@ fix_verify_fun(SslOptsConfig) ->
             end
     end.
 
+make_verify_fun(Module, Function, InitialUserState, UseNewVerifyFun) ->
+    try
+        %% Preload the module: it is required to use
+        %% erlang:function_exported/3.
+        Module:module_info()
+    catch
+        _:Exception ->
+            rabbit_log:error("SSL verify_fun: module ~s missing: ~p~n",
+                             [Module, Exception]),
+            throw({error, {invalid_verify_fun, missing_module}})
+    end,
+    NewForm = erlang:function_exported(Module, Function, 3),
+    OldForm = erlang:function_exported(Module, Function, 1),
+    case {NewForm, OldForm} of
+        {true, _} when UseNewVerifyFun ->
+            %% This verify_fun is supported by Erlang R14B+ (ssl
+            %% 4.0.1 and later).
+            Fun = fun(OtpCert, Event, UserState) ->
+                    Module:Function(OtpCert, Event, UserState)
+            end,
+            {Fun, InitialUserState};
+        {_, true} ->
+            %% This verify_fun is supported by:
+            %%     o  Erlang up-to R13B;
+            %%     o  Erlang R14B+ for undocumented backward
+            %%        compatibility.
+            %%
+            %% InitialUserState is ignored in this case.
+            fun(ErrorList) ->
+                    Module:Function(ErrorList)
+            end;
+        {_, false} when not UseNewVerifyFun ->
+            rabbit_log:error("SSL verify_fun: ~s:~s/1 form required "
+              "for Erlang R13B~n", [Module, Function]),
+            throw({error, {invalid_verify_fun, old_form_required}});
+        _ ->
+            Arity = case UseNewVerifyFun of true -> 3; _ -> 1 end,
+            rabbit_log:error("SSL verify_fun: no ~s:~s/~b exported~n",
+              [Module, Function, Arity]),
+            throw({error, {invalid_verify_fun, function_not_exported}})
+    end.
+
 fix_ssl_protocol_versions(Config) ->
     case application:get_env(rabbit, ssl_allow_poodle_attack) of
         {ok, true} ->
diff --git a/src/rabbit_plugins.erl b/src/rabbit_plugins.erl
index e290fb53..55f7359b 100644
--- a/src/rabbit_plugins.erl
+++ b/src/rabbit_plugins.erl
@@ -90,7 +90,7 @@ list(PluginsDir) ->
     EZs = [{ez, EZ} || EZ <- filelib:wildcard("*.ez", PluginsDir)],
     FreeApps = [{app, App} ||
                    App <- filelib:wildcard("*/ebin/*.app", PluginsDir)],
-    {Plugins, Problems} =
+    {AvailablePlugins, Problems} =
         lists:foldl(fun ({error, EZ, Reason}, {Plugins1, Problems1}) ->
                             {Plugins1, [{EZ, Reason} | Problems1]};
                         (Plugin = #plugin{}, {Plugins1, Problems1}) ->
@@ -102,6 +102,8 @@ list(PluginsDir) ->
         _  -> rabbit_log:warning(
                 "Problem reading some plugins: ~p~n", [Problems])
     end,
+    Plugins = lists:filter(fun(P) -> not plugin_provided_by_otp(P) end,
+                           AvailablePlugins),
     ensure_dependencies(Plugins).
 
 %% @doc Read the list of enabled plugins from the supplied term file.
@@ -132,6 +134,15 @@ dependencies(Reverse, Sources, AllPlugins) ->
     true = digraph:delete(G),
     Dests.
 
+%% For a few known cases, an externally provided plugin can be trusted.
+%% In this special case, it overrides the plugin.
+plugin_provided_by_otp(#plugin{name = eldap}) ->
+    %% eldap was added to Erlang/OTP R15B01 (ERTS 5.9.1). In this case,
+    %% we prefer this version to the plugin.
+    rabbit_misc:version_compare(erlang:system_info(version), "5.9.1", gte);
+plugin_provided_by_otp(_) ->
+    false.
+
 %% Make sure we don't list OTP apps in here, and also that we detect
 %% missing dependencies.
 ensure_dependencies(Plugins) ->
@@ -158,7 +169,7 @@ is_loadable(App) ->
         ok                           -> application:unload(App),
                                         true;
         _                            -> false
-   end.
+    end.
 
 %%----------------------------------------------------------------------------
 
@@ -197,8 +208,27 @@ clean_plugin(Plugin, ExpandDir) ->
     delete_recursively(rabbit_misc:format("~s/~s", [ExpandDir, Plugin])).
 
 prepare_dir_plugin(PluginAppDescPath) ->
-    code:add_path(filename:dirname(PluginAppDescPath)),
-    list_to_atom(filename:basename(PluginAppDescPath, ".app")).
+    PluginEbinDir = filename:dirname(PluginAppDescPath),
+    Plugin = filename:basename(PluginAppDescPath, ".app"),
+    code:add_patha(PluginEbinDir),
+    case filelib:wildcard(PluginEbinDir++ "/*.beam") of
+        [] ->
+            ok;
+        [BeamPath | _] ->
+            Module = list_to_atom(filename:basename(BeamPath, ".beam")),
+            case code:ensure_loaded(Module) of
+                {module, _} ->
+                    ok;
+                {error, badfile} ->
+                    rabbit_log:error("Failed to enable plugin \"~s\": "
+                                     "it may have been built with an "
+                                     "incompatible (more recent?) "
+                                     "version of Erlang~n", [Plugin]),
+                    throw({plugin_built_with_incompatible_erlang, Plugin});
+                Error ->
+                    throw({plugin_module_unloadable, Plugin, Error})
+            end
+    end.
 
 %%----------------------------------------------------------------------------
 
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 0a1c503e..8c545467 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -58,6 +58,11 @@
 
 -define(INFO_KEYS, ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [pid]).
 
+-define(AUTH_NOTIFICATION_INFO_KEYS,
+        [host, vhost, name, peer_host, peer_port, protocol, auth_mechanism,
+         ssl, ssl_protocol, ssl_cipher, peer_cert_issuer, peer_cert_subject,
+         peer_cert_validity]).
+
 -define(IS_RUNNING(State),
         (State#v1.connection_state =:= running orelse
          State#v1.connection_state =:= blocking orelse
@@ -972,7 +977,7 @@ handle_method0(#'connection.open'{virtual_host = VHostPath},
                            helper_sup       = SupPid,
                            sock             = Sock,
                            throttle         = Throttle}) ->
-    ok = rabbit_access_control:check_vhost_access(User, VHostPath),
+    ok = rabbit_access_control:check_vhost_access(User, VHostPath, Sock),
     NewConnection = Connection#connection{vhost = VHostPath},
     ok = send_on_channel0(Sock, #'connection.open_ok'{}, Protocol),
     Conserve = rabbit_alarm:register(self(), {?MODULE, conserve_resources, []}),
@@ -1074,9 +1079,12 @@ auth_phase(Response,
                                        auth_state     = AuthState},
                        sock = Sock}) ->
     case AuthMechanism:handle_response(Response, AuthState) of
-        {refused, Msg, Args} ->
-            auth_fail(Msg, Args, Name, State);
+        {refused, Username, Msg, Args} ->
+            auth_fail(Username, Msg, Args, Name, State);
         {protocol_error, Msg, Args} ->
+            notify_auth_result(none, user_authentication_failure,
+                               [{error, rabbit_misc:format(Msg, Args)}],
+                               State),
             rabbit_misc:protocol_error(syntax_error, Msg, Args);
         {challenge, Challenge, AuthState1} ->
             Secure = #'connection.secure'{challenge = Challenge},
@@ -1085,9 +1093,12 @@ auth_phase(Response,
                                     auth_state = AuthState1}};
         {ok, User = #user{username = Username}} ->
             case rabbit_access_control:check_user_loopback(Username, Sock) of
-                ok          -> ok;
-                not_allowed -> auth_fail("user '~s' can only connect via "
-                                         "localhost", [Username], Name, State)
+                ok ->
+                    notify_auth_result(Username, user_authentication_success,
+                                       [], State);
+                not_allowed ->
+                    auth_fail(Username, "user '~s' can only connect via "
+                              "localhost", [Username], Name, State)
             end,
             Tune = #'connection.tune'{frame_max   = get_env(frame_max),
                                       channel_max = get_env(channel_max),
@@ -1099,11 +1110,15 @@ auth_phase(Response,
     end.
 
 -ifdef(use_specs).
--spec(auth_fail/4 :: (string(), [any()], binary(), #v1{}) -> no_return()).
+-spec(auth_fail/5 ::
+        (rabbit_types:username() | none, string(), [any()], binary(), #v1{}) ->
+           no_return()).
 -endif.
-auth_fail(Msg, Args, AuthName,
+auth_fail(Username, Msg, Args, AuthName,
           State = #v1{connection = #connection{protocol     = Protocol,
                                                capabilities = Capabilities}}) ->
+    notify_auth_result(Username, user_authentication_failure,
+      [{error, rabbit_misc:format(Msg, Args)}], State),
     AmqpError = rabbit_misc:amqp_error(
                   access_refused, "~s login refused: ~s",
                   [AuthName, io_lib:format(Msg, Args)], none),
@@ -1122,6 +1137,16 @@ auth_fail(Msg, Args, AuthName,
     end,
     rabbit_misc:protocol_error(AmqpError).
 
+notify_auth_result(Username, AuthResult, ExtraProps, State) ->
+    EventProps = [{connection_type, network},
+                  {name, case Username of none -> ''; _ -> Username end}] ++
+                 [case Item of
+                      name -> {connection_name, i(name, State)};
+                      _    -> {Item, i(Item, State)}
+                  end || Item <- ?AUTH_NOTIFICATION_INFO_KEYS] ++
+                 ExtraProps,
+    rabbit_event:notify(AuthResult, [P || {_, V} = P <- EventProps, V =/= '']).
+
 %%--------------------------------------------------------------------------
 
 infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
diff --git a/src/rabbit_types.erl b/src/rabbit_types.erl
index ba48867a..039568df 100644
--- a/src/rabbit_types.erl
+++ b/src/rabbit_types.erl
@@ -27,7 +27,7 @@
               vhost/0, ctag/0, amqp_error/0, r/1, r2/2, r3/3, listener/0,
               binding/0, binding_source/0, binding_destination/0,
               amqqueue/0, exchange/0,
-              connection/0, protocol/0, user/0, internal_user/0,
+              connection/0, protocol/0, auth_user/0, user/0, internal_user/0,
               username/0, password/0, password_hash/0,
               ok/1, error/1, ok_or_error/1, ok_or_error2/2, ok_pid_or_error/0,
               channel_exit/0, connection_exit/0, mfargs/0, proc_name/0,
@@ -131,11 +131,15 @@
 
 -type(protocol() :: rabbit_framing:protocol()).
 
+-type(auth_user() ::
+        #auth_user{username :: username(),
+                   tags     :: [atom()],
+                   impl     :: any()}).
+
 -type(user() ::
-        #user{username     :: username(),
-              tags         :: [atom()],
-              auth_backend :: atom(),
-              impl         :: any()}).
+        #user{username       :: username(),
+              tags           :: [atom()],
+              authz_backends :: [{atom(), any()}]}).
 
 -type(internal_user() ::
         #internal_user{username      :: username(),
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index 72bf7855..2ab65459 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -16,7 +16,8 @@
 
 -module(rabbit_upgrade).
 
--export([maybe_upgrade_mnesia/0, maybe_upgrade_local/0]).
+-export([maybe_upgrade_mnesia/0, maybe_upgrade_local/0,
+         nodes_running/1, secondary_upgrade/1]).
 
 -include("rabbit.hrl").
 
@@ -122,6 +123,7 @@ remove_backup() ->
 
 maybe_upgrade_mnesia() ->
     AllNodes = rabbit_mnesia:cluster_nodes(all),
+    ok = rabbit_mnesia_rename:maybe_finish(AllNodes),
     case rabbit_version:upgrades_required(mnesia) of
         {error, starting_from_scratch} ->
             ok;
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index d076b534..1da3de26 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -1198,6 +1198,8 @@ upd_ram_bytes(Sign, MsgStatus, State = #vqstate{ram_bytes = RamBytes}) ->
 
 msg_size(#msg_status{msg_props = #message_properties{size = Size}}) -> Size.
 
+msg_in_ram(#msg_status{msg = Msg}) -> Msg =/= undefined.
+
 remove(AckRequired, MsgStatus = #msg_status {
                       seq_id        = SeqId,
                       msg_id        = MsgId,
@@ -1485,7 +1487,11 @@ publish_alpha(MsgStatus, State) ->
 
 publish_beta(MsgStatus, State) ->
     {MsgStatus1, State1} = maybe_write_to_disk(true, false, MsgStatus, State),
-    {m(trim_msg_status(MsgStatus1)), State1}.
+    MsgStatus2 = m(trim_msg_status(MsgStatus1)),
+    case msg_in_ram(MsgStatus1) andalso not msg_in_ram(MsgStatus2) of
+        true -> {MsgStatus2, upd_ram_bytes(-1, MsgStatus, State1)};
+        _    -> {MsgStatus2, State1}
+    end.
 
 %% Rebuild queue, inserting sequence ids to maintain ordering
 queue_merge(SeqIds, Q, MsgIds, Limit, PubFun, State) ->
@@ -1521,8 +1527,13 @@ delta_merge(SeqIds, Delta, MsgIds, State) ->
                             msg_from_pending_ack(SeqId, State0),
                         {_MsgStatus, State2} =
                             maybe_write_to_disk(true, true, MsgStatus, State1),
+                        State3 =
+                            case msg_in_ram(MsgStatus) of
+                                false -> State2;
+                                true  -> upd_ram_bytes(-1, MsgStatus, State2)
+                            end,
                         {expand_delta(SeqId, Delta0), [MsgId | MsgIds0],
-                         upd_bytes(1, -1, MsgStatus, State2)}
+                         upd_bytes(1, -1, MsgStatus, State3)}
                 end, {Delta, MsgIds, State}, SeqIds).
 
 %% Mostly opposite of record_pending_ack/2
diff --git a/test/src/rabbit_tests.erl b/test/src/rabbit_tests.erl
index ef6b756b..dcbec8f6 100644
--- a/test/src/rabbit_tests.erl
+++ b/test/src/rabbit_tests.erl
@@ -1292,11 +1292,9 @@ test_spawn_remote() ->
     end.
 
 user(Username) ->
-    #user{username     = Username,
-          tags         = [administrator],
-          auth_backend = rabbit_auth_backend_internal,
-          impl         = #internal_user{username = Username,
-                                        tags     = [administrator]}}.
+    #user{username       = Username,
+          tags           = [administrator],
+          authz_backends = [{rabbit_auth_backend_internal, none}]}.
 
 test_confirms() ->
     {_Writer, Ch} = test_spawn(),