Merged bug23442 into default

16 files changed, 509 insertions, 382 deletions

diff --git a/docs/rabbitmq-server.1.xml b/docs/rabbitmq-server.1.xml
index 921da4f1..03e76c79 100644
--- a/docs/rabbitmq-server.1.xml
+++ b/docs/rabbitmq-server.1.xml
@@ -98,18 +98,6 @@ Defaults to 5672.
           </listitem>
         </varlistentry>
 
-        <varlistentry>
-          <term>RABBITMQ_CLUSTER_CONFIG_FILE</term>
-          <listitem>
-            <para>
-Defaults to <filename>/etc/rabbitmq/rabbitmq_cluster.config</filename>. If this file is
-present it is used by the server to auto-configure a RabbitMQ cluster.
-See the <ulink url="http://www.rabbitmq.com/clustering.html">clustering guide</ulink>
-for details.
-            </para>
-          </listitem>
-        </varlistentry>
-
       </variablelist>
     </refsect1>
 
diff --git a/docs/rabbitmq-service.xml b/docs/rabbitmq-service.xml
index 2b416e3e..e95f9889 100644
--- a/docs/rabbitmq-service.xml
+++ b/docs/rabbitmq-service.xml
@@ -193,18 +193,6 @@ manager.
         </varlistentry>
 
         <varlistentry>
-          <term>RABBITMQ_CLUSTER_CONFIG_FILE</term>
-          <listitem>
-            <para>
-If this file is
-present it is used by the server to auto-configure a RabbitMQ cluster.
-See the <ulink url="http://www.rabbitmq.com/clustering.html">clustering guide</ulink>
-for details.
-            </para>
-          </listitem>
-        </varlistentry>
-
-        <varlistentry>
           <term>RABBITMQ_CONSOLE_LOG</term>
           <listitem>
             <para>
diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index 4be09c5a..17d05a99 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -29,4 +29,6 @@
          {default_user_is_admin, true},
          {default_vhost, <<"/">>},
          {default_permissions, [<<".*">>, <<".*">>, <<".*">>]},
+         {cluster_nodes, []},
+         {server_properties, []},
          {collect_statistics, none}]}]}.
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index c1c9bd65..4d4f6fe6 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -63,7 +63,7 @@
 -record(binding, {source, key, destination, args = []}).
 -record(reverse_binding, {destination, key, source, args = []}).
 
--record(listener, {node, protocol, host, port}).
+-record(listener, {node, protocol, host, ip_address, port}).
 
 -record(basic_message, {exchange_name, routing_key, content, guid,
                         is_persistent}).
diff --git a/packaging/macports/Portfile.in b/packaging/macports/Portfile.in
index f30460d3..e37a45b3 100644
--- a/packaging/macports/Portfile.in
+++ b/packaging/macports/Portfile.in
@@ -75,7 +75,7 @@ post-destroot {
 
     reinplace -E "s:(/etc/rabbitmq/rabbitmq.conf):${prefix}\\1:g" \
         ${realsbin}/rabbitmq-env
-    foreach var {CONFIG_FILE CLUSTER_CONFIG_FILE LOG_BASE MNESIA_BASE PIDS_FILE} {
+    foreach var {CONFIG_FILE LOG_BASE MNESIA_BASE PIDS_FILE} {
         reinplace -E "s:^($var)=/:\\1=${prefix}/:" \
             ${realsbin}/rabbitmq-multi \
             ${realsbin}/rabbitmq-server \
diff --git a/scripts/rabbitmq-server b/scripts/rabbitmq-server
index 8e26663a..ef0a3521 100755
--- a/scripts/rabbitmq-server
+++ b/scripts/rabbitmq-server
@@ -35,7 +35,6 @@ NODENAME=rabbit@${HOSTNAME%%.*}
 SERVER_ERL_ARGS="+K true +A30 +P 1048576 \
 -kernel inet_default_listen_options [{nodelay,true}] \
 -kernel inet_default_connect_options [{nodelay,true}]"
-CLUSTER_CONFIG_FILE=/etc/rabbitmq/rabbitmq_cluster.config
 CONFIG_FILE=/etc/rabbitmq/rabbitmq
 LOG_BASE=/var/log/rabbitmq
 MNESIA_BASE=/var/lib/rabbitmq/mnesia
@@ -59,7 +58,6 @@ else
 fi
 [ "x" = "x$RABBITMQ_NODENAME" ] && RABBITMQ_NODENAME=${NODENAME}
 [ "x" = "x$RABBITMQ_SERVER_ERL_ARGS" ] && RABBITMQ_SERVER_ERL_ARGS=${SERVER_ERL_ARGS}
-[ "x" = "x$RABBITMQ_CLUSTER_CONFIG_FILE" ] && RABBITMQ_CLUSTER_CONFIG_FILE=${CLUSTER_CONFIG_FILE}
 [ "x" = "x$RABBITMQ_CONFIG_FILE" ] && RABBITMQ_CONFIG_FILE=${CONFIG_FILE}
 [ "x" = "x$RABBITMQ_LOG_BASE" ] && RABBITMQ_LOG_BASE=${LOG_BASE}
 [ "x" = "x$RABBITMQ_MNESIA_BASE" ] && RABBITMQ_MNESIA_BASE=${MNESIA_BASE}
diff --git a/scripts/rabbitmq-server.bat b/scripts/rabbitmq-server.bat
index 5bcbc6ba..193f1c8a 100644
--- a/scripts/rabbitmq-server.bat
+++ b/scripts/rabbitmq-server.bat
@@ -165,7 +165,6 @@ if not "!RABBITMQ_NODE_IP_ADDRESS!"=="" (
 -os_mon start_disksup false ^

 -os_mon start_memsup false ^

 -mnesia dir \""!RABBITMQ_MNESIA_DIR!"\" ^

-!CLUSTER_CONFIG! ^

 !RABBITMQ_SERVER_START_ARGS! ^

 !STAR!

 

diff --git a/scripts/rabbitmq-service.bat b/scripts/rabbitmq-service.bat
index 4b3961d4..51ef6ecc 100644
--- a/scripts/rabbitmq-service.bat
+++ b/scripts/rabbitmq-service.bat
@@ -232,7 +232,6 @@ set ERLANG_SERVICE_ARGUMENTS= ^
 -os_mon start_disksup false ^

 -os_mon start_memsup false ^

 -mnesia dir \""!RABBITMQ_MNESIA_DIR!"\" ^

-!CLUSTER_CONFIG! ^

 !RABBITMQ_SERVER_START_ARGS! ^

 !STAR!

 

diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 577d206d..8de2f0d6 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -340,10 +340,8 @@ read_cluster_nodes_config() ->
     case rabbit_misc:read_term_file(FileName) of
         {ok, [ClusterNodes]} -> ClusterNodes;
         {error, enoent} ->
-            case application:get_env(cluster_nodes) of
-                undefined -> [];
-                {ok, ClusterNodes} -> ClusterNodes
-            end;
+            {ok, ClusterNodes} = application:get_env(rabbit, cluster_nodes),
+            ClusterNodes;
         {error, Reason} ->
             throw({error, {cannot_read_cluster_nodes_config,
                            FileName, Reason}})
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index b9f3e1a3..24b017b5 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -34,8 +34,9 @@
 -behaviour(gen_server2).
 
 -export([start_link/4, successfully_recovered_state/1,
-         client_init/2, client_terminate/2, client_delete_and_terminate/3,
-         write/4, read/3, contains/2, remove/2, release/2, sync/3]).
+         client_init/2, client_terminate/1, client_delete_and_terminate/1,
+         client_ref/1,
+         write/3, read/2, contains/2, remove/2, release/2, sync/3]).
 
 -export([sync/1, set_maximum_since_use/2,
          has_readers/2, combine_files/3, delete_file/2]). %% internal
@@ -86,7 +87,9 @@
          }).
 
 -record(client_msstate,
-        { file_handle_cache,
+        { server,
+          client_ref,
+          file_handle_cache,
           index_state,
           index_module,
           dir,
@@ -122,8 +125,11 @@
                                 }).
 
 -type(server() :: pid() | atom()).
+-type(client_ref() :: binary()).
 -type(file_num() :: non_neg_integer()).
 -type(client_msstate() :: #client_msstate {
+                      server             :: server(),
+                      client_ref         :: client_ref(),
                       file_handle_cache  :: dict:dictionary(),
                       index_state        :: any(),
                       index_module       :: atom(),
@@ -141,18 +147,18 @@
         (atom(), file:filename(), [binary()] | 'undefined',
          startup_fun_state()) -> rabbit_types:ok_pid_or_error()).
 -spec(successfully_recovered_state/1 :: (server()) -> boolean()).
--spec(client_init/2 :: (server(), binary()) -> client_msstate()).
--spec(client_terminate/2 :: (client_msstate(), server()) -> 'ok').
--spec(client_delete_and_terminate/3 ::
-        (client_msstate(), server(), binary()) -> 'ok').
--spec(write/4 :: (server(), rabbit_guid:guid(), msg(), client_msstate()) ->
-                      rabbit_types:ok(client_msstate())).
--spec(read/3 :: (server(), rabbit_guid:guid(), client_msstate()) ->
-                     {rabbit_types:ok(msg()) | 'not_found', client_msstate()}).
--spec(contains/2 :: (server(), rabbit_guid:guid()) -> boolean()).
--spec(remove/2 :: (server(), [rabbit_guid:guid()]) -> 'ok').
--spec(release/2 :: (server(), [rabbit_guid:guid()]) -> 'ok').
--spec(sync/3 :: (server(), [rabbit_guid:guid()], fun (() -> any())) -> 'ok').
+-spec(client_init/2 :: (server(), client_ref()) -> client_msstate()).
+-spec(client_terminate/1 :: (client_msstate()) -> 'ok').
+-spec(client_delete_and_terminate/1 :: (client_msstate()) -> 'ok').
+-spec(client_ref/1 :: (client_msstate()) -> client_ref()).
+-spec(write/3 :: (rabbit_guid:guid(), msg(), client_msstate()) -> 'ok').
+-spec(read/2 :: (rabbit_guid:guid(), client_msstate()) ->
+             {rabbit_types:ok(msg()) | 'not_found', client_msstate()}).
+-spec(contains/2 :: (rabbit_guid:guid(), client_msstate()) -> boolean()).
+-spec(remove/2 :: ([rabbit_guid:guid()], client_msstate()) -> 'ok').
+-spec(release/2 :: ([rabbit_guid:guid()], client_msstate()) -> 'ok').
+-spec(sync/3 :: ([rabbit_guid:guid()], fun (() -> any()), client_msstate()) ->
+             'ok').
 
 -spec(sync/1 :: (server()) -> 'ok').
 -spec(set_maximum_since_use/2 :: (server(), non_neg_integer()) -> 'ok').
@@ -332,7 +338,9 @@ client_init(Server, Ref) ->
     {IState, IModule, Dir, GCPid,
      FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts} =
         gen_server2:call(Server, {new_client_state, Ref}, infinity),
-    #client_msstate { file_handle_cache  = dict:new(),
+    #client_msstate { server             = Server,
+                      client_ref         = Ref,
+                      file_handle_cache  = dict:new(),
                       index_state        = IState,
                       index_module       = IModule,
                       dir                = Dir,
@@ -342,20 +350,22 @@ client_init(Server, Ref) ->
                       dedup_cache_ets    = DedupCacheEts,
                       cur_file_cache_ets = CurFileCacheEts }.
 
-client_terminate(CState, Server) ->
+client_terminate(CState) ->
     close_all_handles(CState),
-    ok = gen_server2:call(Server, client_terminate, infinity).
+    ok = server_call(CState, client_terminate).
 
-client_delete_and_terminate(CState, Server, Ref) ->
+client_delete_and_terminate(CState = #client_msstate { client_ref = Ref }) ->
     close_all_handles(CState),
-    ok = gen_server2:cast(Server, {client_delete, Ref}).
+    ok = server_cast(CState, {client_delete, Ref}).
 
-write(Server, Guid, Msg,
+client_ref(#client_msstate { client_ref = Ref }) -> Ref.
+
+write(Guid, Msg,
       CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts }) ->
     ok = update_msg_cache(CurFileCacheEts, Guid, Msg),
-    {gen_server2:cast(Server, {write, Guid}), CState}.
+    ok = server_cast(CState, {write, Guid}).
 
-read(Server, Guid,
+read(Guid,
      CState = #client_msstate { dedup_cache_ets    = DedupCacheEts,
                                 cur_file_cache_ets = CurFileCacheEts }) ->
     %% 1. Check the dedup cache
@@ -364,13 +374,12 @@ read(Server, Guid,
             %% 2. Check the cur file cache
             case ets:lookup(CurFileCacheEts, Guid) of
                 [] ->
-                    Defer = fun() -> {gen_server2:call(
-                                        Server, {read, Guid}, infinity),
-                                      CState} end,
+                    Defer = fun() ->
+                                    {server_call(CState, {read, Guid}), CState}
+                            end,
                     case index_lookup_positive_ref_count(Guid, CState) of
                         not_found   -> Defer();
-                        MsgLocation -> client_read1(Server, MsgLocation, Defer,
-                                                    CState)
+                        MsgLocation -> client_read1(MsgLocation, Defer, CState)
                     end;
                 [{Guid, Msg, _CacheRefCount}] ->
                     %% Although we've found it, we don't know the
@@ -381,12 +390,12 @@ read(Server, Guid,
             {{ok, Msg}, CState}
     end.
 
-contains(Server, Guid) -> gen_server2:call(Server, {contains, Guid}, infinity).
-remove(_Server, [])    -> ok;
-remove(Server, Guids)  -> gen_server2:cast(Server, {remove, Guids}).
-release(_Server, [])   -> ok;
-release(Server, Guids) -> gen_server2:cast(Server, {release, Guids}).
-sync(Server, Guids, K) -> gen_server2:cast(Server, {sync, Guids, K}).
+contains(Guid, CState) -> server_call(CState, {contains, Guid}).
+remove([],    _CState) -> ok;
+remove(Guids,  CState) -> server_cast(CState, {remove, Guids}).
+release([],   _CState) -> ok;
+release(Guids, CState) -> server_cast(CState, {release, Guids}).
+sync(Guids, K, CState) -> server_cast(CState, {sync, Guids, K}).
 
 sync(Server) ->
     gen_server2:cast(Server, sync).
@@ -398,18 +407,22 @@ set_maximum_since_use(Server, Age) ->
 %% Client-side-only helpers
 %%----------------------------------------------------------------------------
 
-client_read1(Server,
-             #msg_location { guid = Guid, file = File } = MsgLocation,
-             Defer,
+server_call(#client_msstate { server = Server }, Msg) ->
+    gen_server2:call(Server, Msg, infinity).
+
+server_cast(#client_msstate { server = Server }, Msg) ->
+    gen_server2:cast(Server, Msg).
+
+client_read1(#msg_location { guid = Guid, file = File } = MsgLocation, Defer,
              CState = #client_msstate { file_summary_ets = FileSummaryEts }) ->
     case ets:lookup(FileSummaryEts, File) of
         [] -> %% File has been GC'd and no longer exists. Go around again.
-            read(Server, Guid, CState);
+            read(Guid, CState);
         [#file_summary { locked = Locked, right = Right }] ->
-            client_read2(Server, Locked, Right, MsgLocation, Defer, CState)
+            client_read2(Locked, Right, MsgLocation, Defer, CState)
     end.
 
-client_read2(_Server, false, undefined, _MsgLocation, Defer, _CState) ->
+client_read2(false, undefined, _MsgLocation, Defer, _CState) ->
     %% Although we've already checked both caches and not found the
     %% message there, the message is apparently in the
     %% current_file. We can only arrive here if we are trying to read
@@ -420,12 +433,12 @@ client_read2(_Server, false, undefined, _MsgLocation, Defer, _CState) ->
     %% contents of the current file, thus reads from the current file
     %% will end up here and will need to be deferred.
     Defer();
-client_read2(_Server, true, _Right, _MsgLocation, Defer, _CState) ->
+client_read2(true, _Right, _MsgLocation, Defer, _CState) ->
     %% Of course, in the mean time, the GC could have run and our msg
     %% is actually in a different file, unlocked. However, defering is
     %% the safest and simplest thing to do.
     Defer();
-client_read2(Server, false, _Right,
+client_read2(false, _Right,
              MsgLocation = #msg_location { guid = Guid, file = File },
              Defer,
              CState = #client_msstate { file_summary_ets = FileSummaryEts }) ->
@@ -434,10 +447,10 @@ client_read2(Server, false, _Right,
     %% finished.
     safe_ets_update_counter(
       FileSummaryEts, File, {#file_summary.readers, +1},
-      fun (_) -> client_read3(Server, MsgLocation, Defer, CState) end,
-      fun () -> read(Server, Guid, CState) end).
+      fun (_) -> client_read3(MsgLocation, Defer, CState) end,
+      fun () -> read(Guid, CState) end).
 
-client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
+client_read3(#msg_location { guid = Guid, file = File }, Defer,
              CState = #client_msstate { file_handles_ets = FileHandlesEts,
                                         file_summary_ets = FileSummaryEts,
                                         dedup_cache_ets  = DedupCacheEts,
@@ -461,7 +474,7 @@ client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
     %% too).
     case ets:lookup(FileSummaryEts, File) of
         [] -> %% GC has deleted our file, just go round again.
-            read(Server, Guid, CState);
+            read(Guid, CState);
         [#file_summary { locked = true }] ->
             %% If we get a badarg here, then the GC has finished and
             %% deleted our file. Try going around again. Otherwise,
@@ -472,7 +485,7 @@ client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
             %% unlocks the dest)
             try Release(),
                 Defer()
-            catch error:badarg -> read(Server, Guid, CState)
+            catch error:badarg -> read(Guid, CState)
             end;
         [#file_summary { locked = false }] ->
             %% Ok, we're definitely safe to continue - a GC involving
@@ -497,7 +510,7 @@ client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
                     {{ok, Msg}, CState2};
                 #msg_location {} = MsgLocation -> %% different file!
                     Release(), %% this MUST NOT fail with badarg
-                    client_read1(Server, MsgLocation, Defer, CState);
+                    client_read1(MsgLocation, Defer, CState);
                 not_found -> %% it seems not to exist. Defer, just to be sure.
                     try Release() %% this can badarg, same as locked case, above
                     catch error:badarg -> ok
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index fd04109f..03a9b386 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -189,11 +189,15 @@ stop_tcp_listener(Host, Port) ->
     ok.
 
 tcp_listener_started(Protocol, IPAddress, Port) ->
+    %% We need the ip to distinguish e.g. 0.0.0.0 and 127.0.0.1
+    %% We need the host so we can distinguish multiple instances of the above
+    %% in a cluster.
     ok = mnesia:dirty_write(
            rabbit_listener,
            #listener{node = node(),
                      protocol = Protocol,
                      host = tcp_host(IPAddress),
+                     ip_address = IPAddress,
                      port = Port}).
 
 tcp_listener_stopped(Protocol, IPAddress, Port) ->
@@ -202,6 +206,7 @@ tcp_listener_stopped(Protocol, IPAddress, Port) ->
            #listener{node = node(),
                      protocol = Protocol,
                      host = tcp_host(IPAddress),
+                     ip_address = IPAddress,
                      port = Port}).
 
 active_listeners() ->
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index 1b837128..28d0b47d 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -31,8 +31,9 @@
 
 -module(rabbit_queue_index).
 
--export([init/4, terminate/2, delete_and_terminate/1, publish/5,
-         deliver/2, ack/2, sync/2, flush/1, read/3,
+-export([init/1, shutdown_terms/1, recover/4,
+         terminate/2, delete_and_terminate/1,
+         publish/5, deliver/2, ack/2, sync/2, flush/1, read/3,
          next_segment_boundary/1, bounds/1, recover/1]).
 
 -define(CLEAN_FILENAME, "clean.dot").
@@ -199,10 +200,13 @@
 -type(startup_fun_state() ::
         {(fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A})),
          A}).
+-type(shutdown_terms() :: [any()]).
 
--spec(init/4 :: (rabbit_amqqueue:name(), boolean(), boolean(),
-                 fun ((rabbit_guid:guid()) -> boolean())) ->
-             {'undefined' | non_neg_integer(), [any()], qistate()}).
+-spec(init/1 :: (rabbit_amqqueue:name()) -> qistate()).
+-spec(shutdown_terms/1 :: (rabbit_amqqueue:name()) -> shutdown_terms()).
+-spec(recover/4 :: (rabbit_amqqueue:name(), shutdown_terms(), boolean(),
+                    fun ((rabbit_guid:guid()) -> boolean())) ->
+             {'undefined' | non_neg_integer(), qistate()}).
 -spec(terminate/2 :: ([any()], qistate()) -> qistate()).
 -spec(delete_and_terminate/1 :: (qistate()) -> qistate()).
 -spec(publish/5 :: (rabbit_guid:guid(), seq_id(),
@@ -229,25 +233,26 @@
 %% public API
 %%----------------------------------------------------------------------------
 
-init(Name, false, _MsgStoreRecovered, _ContainsCheckFun) ->
+init(Name) ->
     State = #qistate { dir = Dir } = blank_state(Name),
     false = filelib:is_file(Dir), %% is_file == is file or dir
-    {0, [], State};
+    State.
+
+shutdown_terms(Name) ->
+    #qistate { dir = Dir } = blank_state(Name),
+    case read_shutdown_terms(Dir) of
+        {error, _}   -> [];
+        {ok, Terms1} -> Terms1
+    end.
 
-init(Name, true, MsgStoreRecovered, ContainsCheckFun) ->
+recover(Name, Terms, MsgStoreRecovered, ContainsCheckFun) ->
     State = #qistate { dir = Dir } = blank_state(Name),
-    Terms = case read_shutdown_terms(Dir) of
-                {error, _}   -> [];
-                {ok, Terms1} -> Terms1
-            end,
     CleanShutdown = detect_clean_shutdown(Dir),
-    {Count, State1} =
-        case CleanShutdown andalso MsgStoreRecovered of
-            true  -> RecoveredCounts = proplists:get_value(segments, Terms, []),
-                     init_clean(RecoveredCounts, State);
-            false -> init_dirty(CleanShutdown, ContainsCheckFun, State)
-        end,
-    {Count, Terms, State1}.
+    case CleanShutdown andalso MsgStoreRecovered of
+        true  -> RecoveredCounts = proplists:get_value(segments, Terms, []),
+                 init_clean(RecoveredCounts, State);
+        false -> init_dirty(CleanShutdown, ContainsCheckFun, State)
+    end.
 
 terminate(Terms, State) ->
     {SegmentCounts, State1 = #qistate { dir = Dir }} = terminate(State),
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 29004bd5..58d29008 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -235,12 +235,29 @@ conserve_memory(Pid, Conserve) ->
 server_properties() ->
     {ok, Product} = application:get_key(rabbit, id),
     {ok, Version} = application:get_key(rabbit, vsn),
-    [{list_to_binary(K), longstr, list_to_binary(V)} ||
-        {K, V} <- [{"product",     Product},
-                   {"version",     Version},
-                   {"platform",    "Erlang/OTP"},
-                   {"copyright",   ?COPYRIGHT_MESSAGE},
-                   {"information", ?INFORMATION_MESSAGE}]].
+
+    %% Get any configuration-specified server properties
+    {ok, RawConfigServerProps} = application:get_env(rabbit,
+                                                     server_properties),
+
+    %% Normalize the simplifed (2-tuple) and unsimplified (3-tuple) forms
+    %% from the config and merge them with the generated built-in properties
+    NormalizedConfigServerProps =
+        [case X of
+             {KeyAtom, Value} -> {atom_to_binary(KeyAtom, latin1),
+                                  longstr,
+                                  list_to_binary(Value)};
+             {BinKey, Type, Value} -> {BinKey, Type, Value}
+         end || X <- RawConfigServerProps ++
+                    [{product,     Product},
+                     {version,     Version},
+                     {platform,    "Erlang/OTP"},
+                     {copyright,   ?COPYRIGHT_MESSAGE},
+                     {information, ?INFORMATION_MESSAGE}]],
+
+    %% Filter duplicated properties in favor of config file provided values
+    lists:usort(fun ({K1,_,_}, {K2,_,_}) -> K1 =< K2 end,
+                NormalizedConfigServerProps).
 
 inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F).
 
diff --git a/src/rabbit_ssl.erl b/src/rabbit_ssl.erl
index be451af6..4335dd2e 100644
--- a/src/rabbit_ssl.erl
+++ b/src/rabbit_ssl.erl
@@ -34,7 +34,6 @@
 -include("rabbit.hrl").
 
 -include_lib("public_key/include/public_key.hrl").
--include_lib("ssl/src/ssl_int.hrl").
 
 -export([peer_cert_issuer/1, peer_cert_subject/1, peer_cert_validity/1]).
 
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index f2a65eeb..71b23e01 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -73,6 +73,7 @@ all_tests() ->
     passed = test_user_management(),
     passed = test_server_status(),
     passed = maybe_run_cluster_dependent_tests(),
+    passed = test_configurable_server_properties(),
     passed.
 
 maybe_run_cluster_dependent_tests() ->
@@ -1426,17 +1427,17 @@ restart_msg_store_empty() ->
 guid_bin(X) ->
     erlang:md5(term_to_binary(X)).
 
-msg_store_contains(Atom, Guids) ->
+msg_store_contains(Atom, Guids, MSCState) ->
     Atom = lists:foldl(
              fun (Guid, Atom1) when Atom1 =:= Atom ->
-                     rabbit_msg_store:contains(?PERSISTENT_MSG_STORE, Guid) end,
+                     rabbit_msg_store:contains(Guid, MSCState) end,
              Atom, Guids).
 
-msg_store_sync(Guids) ->
+msg_store_sync(Guids, MSCState) ->
     Ref = make_ref(),
     Self = self(),
-    ok = rabbit_msg_store:sync(?PERSISTENT_MSG_STORE, Guids,
-                               fun () -> Self ! {sync, Ref} end),
+    ok = rabbit_msg_store:sync(Guids, fun () -> Self ! {sync, Ref} end,
+                               MSCState),
     receive
         {sync, Ref} -> ok
     after
@@ -1448,55 +1449,64 @@ msg_store_sync(Guids) ->
 msg_store_read(Guids, MSCState) ->
     lists:foldl(fun (Guid, MSCStateM) ->
                         {{ok, Guid}, MSCStateN} = rabbit_msg_store:read(
-                                                    ?PERSISTENT_MSG_STORE,
                                                     Guid, MSCStateM),
                         MSCStateN
                 end, MSCState, Guids).
 
 msg_store_write(Guids, MSCState) ->
-    lists:foldl(fun (Guid, {ok, MSCStateN}) ->
-                        rabbit_msg_store:write(?PERSISTENT_MSG_STORE,
-                                               Guid, Guid, MSCStateN)
-                end, {ok, MSCState}, Guids).
+    ok = lists:foldl(
+           fun (Guid, ok) -> rabbit_msg_store:write(Guid, Guid, MSCState) end,
+           ok, Guids).
+
+msg_store_remove(Guids, MSCState) ->
+    rabbit_msg_store:remove(Guids, MSCState).
+
+msg_store_remove(MsgStore, Ref, Guids) ->
+    with_msg_store_client(MsgStore, Ref,
+                          fun (MSCStateM) ->
+                                  ok = msg_store_remove(Guids, MSCStateM),
+                                  MSCStateM
+                          end).
 
-msg_store_remove(Guids) ->
-    rabbit_msg_store:remove(?PERSISTENT_MSG_STORE, Guids).
+with_msg_store_client(MsgStore, Ref, Fun) ->
+    rabbit_msg_store:client_terminate(
+      Fun(rabbit_msg_store:client_init(MsgStore, Ref))).
 
 foreach_with_msg_store_client(MsgStore, Ref, Fun, L) ->
     rabbit_msg_store:client_terminate(
-      lists:foldl(fun (Guid, MSCState) -> Fun(Guid, MsgStore, MSCState) end,
-                  rabbit_msg_store:client_init(MsgStore, Ref), L), MsgStore).
+      lists:foldl(fun (Guid, MSCState) -> Fun(Guid, MSCState) end,
+                  rabbit_msg_store:client_init(MsgStore, Ref), L)).
 
 test_msg_store() ->
     restart_msg_store_empty(),
     Self = self(),
     Guids = [guid_bin(M) || M <- lists:seq(1,100)],
     {Guids1stHalf, Guids2ndHalf} = lists:split(50, Guids),
-    %% check we don't contain any of the msgs we're about to publish
-    false = msg_store_contains(false, Guids),
     Ref = rabbit_guid:guid(),
     MSCState = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
+    %% check we don't contain any of the msgs we're about to publish
+    false = msg_store_contains(false, Guids, MSCState),
     %% publish the first half
-    {ok, MSCState1} = msg_store_write(Guids1stHalf, MSCState),
+    ok = msg_store_write(Guids1stHalf, MSCState),
     %% sync on the first half
-    ok = msg_store_sync(Guids1stHalf),
+    ok = msg_store_sync(Guids1stHalf, MSCState),
     %% publish the second half
-    {ok, MSCState2} = msg_store_write(Guids2ndHalf, MSCState1),
+    ok = msg_store_write(Guids2ndHalf, MSCState),
     %% sync on the first half again - the msg_store will be dirty, but
     %% we won't need the fsync
-    ok = msg_store_sync(Guids1stHalf),
+    ok = msg_store_sync(Guids1stHalf, MSCState),
     %% check they're all in there
-    true = msg_store_contains(true, Guids),
+    true = msg_store_contains(true, Guids, MSCState),
     %% publish the latter half twice so we hit the caching and ref count code
-    {ok, MSCState3} = msg_store_write(Guids2ndHalf, MSCState2),
+    ok = msg_store_write(Guids2ndHalf, MSCState),
     %% check they're still all in there
-    true = msg_store_contains(true, Guids),
+    true = msg_store_contains(true, Guids, MSCState),
     %% sync on the 2nd half, but do lots of individual syncs to try
     %% and cause coalescing to happen
     ok = lists:foldl(
            fun (Guid, ok) -> rabbit_msg_store:sync(
-                                ?PERSISTENT_MSG_STORE,
-                                [Guid], fun () -> Self ! {sync, Guid} end)
+                               [Guid], fun () -> Self ! {sync, Guid} end,
+                               MSCState)
            end, ok, Guids2ndHalf),
     lists:foldl(
       fun(Guid, ok) ->
@@ -1511,24 +1521,24 @@ test_msg_store() ->
       end, ok, Guids2ndHalf),
     %% it's very likely we're not dirty here, so the 1st half sync
     %% should hit a different code path
-    ok = msg_store_sync(Guids1stHalf),
+    ok = msg_store_sync(Guids1stHalf, MSCState),
     %% read them all
-    MSCState4 = msg_store_read(Guids, MSCState3),
+    MSCState1 = msg_store_read(Guids, MSCState),
     %% read them all again - this will hit the cache, not disk
-    MSCState5 = msg_store_read(Guids, MSCState4),
+    MSCState2 = msg_store_read(Guids, MSCState1),
     %% remove them all
-    ok = rabbit_msg_store:remove(?PERSISTENT_MSG_STORE, Guids),
+    ok = rabbit_msg_store:remove(Guids, MSCState2),
     %% check first half doesn't exist
-    false = msg_store_contains(false, Guids1stHalf),
+    false = msg_store_contains(false, Guids1stHalf, MSCState2),
     %% check second half does exist
-    true = msg_store_contains(true, Guids2ndHalf),
+    true = msg_store_contains(true, Guids2ndHalf, MSCState2),
     %% read the second half again
-    MSCState6 = msg_store_read(Guids2ndHalf, MSCState5),
+    MSCState3 = msg_store_read(Guids2ndHalf, MSCState2),
     %% release the second half, just for fun (aka code coverage)
-    ok = rabbit_msg_store:release(?PERSISTENT_MSG_STORE, Guids2ndHalf),
+    ok = rabbit_msg_store:release(Guids2ndHalf, MSCState3),
     %% read the second half again, just for fun (aka code coverage)
-    MSCState7 = msg_store_read(Guids2ndHalf, MSCState6),
-    ok = rabbit_msg_store:client_terminate(MSCState7, ?PERSISTENT_MSG_STORE),
+    MSCState4 = msg_store_read(Guids2ndHalf, MSCState3),
+    ok = rabbit_msg_store:client_terminate(MSCState4),
     %% stop and restart, preserving every other msg in 2nd half
     ok = rabbit_variable_queue:stop_msg_store(),
     ok = rabbit_variable_queue:start_msg_store(
@@ -1539,22 +1549,26 @@ test_msg_store() ->
                     ([Guid|GuidsTail]) ->
                         {Guid, 0, GuidsTail}
                 end, Guids2ndHalf}),
+    MSCState5 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
     %% check we have the right msgs left
     lists:foldl(
       fun (Guid, Bool) ->
-              not(Bool = rabbit_msg_store:contains(?PERSISTENT_MSG_STORE, Guid))
+              not(Bool = rabbit_msg_store:contains(Guid, MSCState5))
       end, false, Guids2ndHalf),
+    ok = rabbit_msg_store:client_terminate(MSCState5),
     %% restart empty
     restart_msg_store_empty(),
+    MSCState6 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
     %% check we don't contain any of the msgs
-    false = msg_store_contains(false, Guids),
+    false = msg_store_contains(false, Guids, MSCState6),
     %% publish the first half again
-    MSCState8 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
-    {ok, MSCState9} = msg_store_write(Guids1stHalf, MSCState8),
+    ok = msg_store_write(Guids1stHalf, MSCState6),
     %% this should force some sort of sync internally otherwise misread
     ok = rabbit_msg_store:client_terminate(
-           msg_store_read(Guids1stHalf, MSCState9), ?PERSISTENT_MSG_STORE),
-    ok = rabbit_msg_store:remove(?PERSISTENT_MSG_STORE, Guids1stHalf),
+           msg_store_read(Guids1stHalf, MSCState6)),
+    MSCState7 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
+    ok = rabbit_msg_store:remove(Guids1stHalf, MSCState7),
+    ok = rabbit_msg_store:client_terminate(MSCState7),
     %% restart empty
     restart_msg_store_empty(), %% now safe to reuse guids
     %% push a lot of msgs in... at least 100 files worth
@@ -1563,31 +1577,39 @@ test_msg_store() ->
     BigCount = trunc(100 * FileSize / (PayloadSizeBits div 8)),
     GuidsBig = [guid_bin(X) || X <- lists:seq(1, BigCount)],
     Payload = << 0:PayloadSizeBits >>,
-    ok = foreach_with_msg_store_client(
+    ok = with_msg_store_client(
            ?PERSISTENT_MSG_STORE, Ref,
-           fun (Guid, MsgStore, MSCStateM) ->
-                   {ok, MSCStateN} = rabbit_msg_store:write(
-                                       MsgStore, Guid, Payload, MSCStateM),
-                   MSCStateN
-           end, GuidsBig),
+           fun (MSCStateM) ->
+                   [ok = rabbit_msg_store:write(Guid, Payload, MSCStateM) ||
+                       Guid <- GuidsBig],
+                   MSCStateM
+           end),
     %% now read them to ensure we hit the fast client-side reading
     ok = foreach_with_msg_store_client(
            ?PERSISTENT_MSG_STORE, Ref,
-           fun (Guid, MsgStore, MSCStateM) ->
+           fun (Guid, MSCStateM) ->
                    {{ok, Payload}, MSCStateN} = rabbit_msg_store:read(
-                                                  MsgStore, Guid, MSCStateM),
+                                                  Guid, MSCStateM),
                    MSCStateN
            end, GuidsBig),
     %% .., then 3s by 1...
-    ok = msg_store_remove([guid_bin(X) || X <- lists:seq(BigCount, 1, -3)]),
+    ok = msg_store_remove(?PERSISTENT_MSG_STORE, Ref,
+                          [guid_bin(X) || X <- lists:seq(BigCount, 1, -3)]),
     %% .., then remove 3s by 2, from the young end first. This hits
     %% GC (under 50% good data left, but no empty files. Must GC).
-    ok = msg_store_remove([guid_bin(X) || X <- lists:seq(BigCount-1, 1, -3)]),
+    ok = msg_store_remove(?PERSISTENT_MSG_STORE, Ref,
+                          [guid_bin(X) || X <- lists:seq(BigCount-1, 1, -3)]),
     %% .., then remove 3s by 3, from the young end first. This hits
     %% GC...
-    ok = msg_store_remove([guid_bin(X) || X <- lists:seq(BigCount-2, 1, -3)]),
+    ok = msg_store_remove(?PERSISTENT_MSG_STORE, Ref,
+                          [guid_bin(X) || X <- lists:seq(BigCount-2, 1, -3)]),
     %% ensure empty
-    false = msg_store_contains(false, GuidsBig),
+    ok = with_msg_store_client(
+           ?PERSISTENT_MSG_STORE, Ref,
+           fun (MSCStateM) ->
+                   false = msg_store_contains(false, GuidsBig, MSCStateM),
+                   MSCStateM
+           end),
     %% restart empty
     restart_msg_store_empty(),
     passed.
@@ -1599,11 +1621,18 @@ test_queue() ->
     queue_name(<<"test">>).
 
 init_test_queue() ->
-    rabbit_queue_index:init(
-      test_queue(), true, false,
-      fun (Guid) ->
-              rabbit_msg_store:contains(?PERSISTENT_MSG_STORE, Guid)
-      end).
+    TestQueue = test_queue(),
+    Terms = rabbit_queue_index:shutdown_terms(TestQueue),
+    PRef = proplists:get_value(persistent_ref, Terms, rabbit_guid:guid()),
+    PersistentClient = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE,
+                                                    PRef),
+    Res = rabbit_queue_index:recover(
+            TestQueue, Terms, false,
+            fun (Guid) ->
+                    rabbit_msg_store:contains(Guid, PersistentClient)
+            end),
+    ok = rabbit_msg_store:client_delete_and_terminate(PersistentClient),
+    Res.
 
 restart_test_queue(Qi) ->
     _ = rabbit_queue_index:terminate([], Qi),
@@ -1614,13 +1643,13 @@ restart_test_queue(Qi) ->
 empty_test_queue() ->
     ok = rabbit_variable_queue:stop(),
     ok = rabbit_variable_queue:start([]),
-    {0, _Terms, Qi} = init_test_queue(),
+    {0, Qi} = init_test_queue(),
     _ = rabbit_queue_index:delete_and_terminate(Qi),
     ok.
 
 with_empty_test_queue(Fun) ->
     ok = empty_test_queue(),
-    {0, _Terms, Qi} = init_test_queue(),
+    {0, Qi} = init_test_queue(),
     rabbit_queue_index:delete_and_terminate(Fun(Qi)).
 
 queue_index_publish(SeqIds, Persistent, Qi) ->
@@ -1629,18 +1658,17 @@ queue_index_publish(SeqIds, Persistent, Qi) ->
                    true  -> ?PERSISTENT_MSG_STORE;
                    false -> ?TRANSIENT_MSG_STORE
                end,
-    {A, B, MSCStateEnd} =
+    MSCState = rabbit_msg_store:client_init(MsgStore, Ref),
+    {A, B} =
         lists:foldl(
-          fun (SeqId, {QiN, SeqIdsGuidsAcc, MSCStateN}) ->
+          fun (SeqId, {QiN, SeqIdsGuidsAcc}) ->
                   Guid = rabbit_guid:guid(),
                   QiM = rabbit_queue_index:publish(
                           Guid, SeqId, #message_properties{}, Persistent, QiN),
-                  {ok, MSCStateM} = rabbit_msg_store:write(MsgStore, Guid,
-                                                           Guid, MSCStateN),
-                  {QiM, [{SeqId, Guid} | SeqIdsGuidsAcc], MSCStateM}
-          end, {Qi, [], rabbit_msg_store:client_init(MsgStore, Ref)}, SeqIds),
-    ok = rabbit_msg_store:client_delete_and_terminate(
-           MSCStateEnd, MsgStore, Ref),
+                  ok = rabbit_msg_store:write(Guid, Guid, MSCState),
+                  {QiM, [{SeqId, Guid} | SeqIdsGuidsAcc]}
+          end, {Qi, []}, SeqIds),
+    ok = rabbit_msg_store:client_delete_and_terminate(MSCState),
     {A, B}.
 
 verify_read_with_published(_Delivered, _Persistent, [], _) ->
@@ -1686,7 +1714,7 @@ test_queue_index() ->
               ok = verify_read_with_published(false, false, ReadA,
                                               lists:reverse(SeqIdsGuidsA)),
               %% should get length back as 0, as all the msgs were transient
-              {0, _Terms1, Qi6} = restart_test_queue(Qi4),
+              {0, Qi6} = restart_test_queue(Qi4),
               {0, 0, Qi7} = rabbit_queue_index:bounds(Qi6),
               {Qi8, SeqIdsGuidsB} = queue_index_publish(SeqIdsB, true, Qi7),
               {0, TwoSegs, Qi9} = rabbit_queue_index:bounds(Qi8),
@@ -1695,7 +1723,7 @@ test_queue_index() ->
                                               lists:reverse(SeqIdsGuidsB)),
               %% should get length back as MostOfASegment
               LenB = length(SeqIdsB),
-              {LenB, _Terms2, Qi12} = restart_test_queue(Qi10),
+              {LenB, Qi12} = restart_test_queue(Qi10),
               {0, TwoSegs, Qi13} = rabbit_queue_index:bounds(Qi12),
               Qi14 = rabbit_queue_index:deliver(SeqIdsB, Qi13),
               {ReadC, Qi15} = rabbit_queue_index:read(0, SegmentSize, Qi14),
@@ -1707,7 +1735,7 @@ test_queue_index() ->
               {0, 0, Qi18} = rabbit_queue_index:bounds(Qi17),
               %% should get length back as 0 because all persistent
               %% msgs have been acked
-              {0, _Terms3, Qi19} = restart_test_queue(Qi18),
+              {0, Qi19} = restart_test_queue(Qi18),
               Qi19
       end),
 
@@ -1779,11 +1807,11 @@ test_queue_index() ->
                                                          true, Qi0),
               Qi2 = rabbit_queue_index:deliver([0,1,4], Qi1),
               Qi3 = rabbit_queue_index:ack([0], Qi2),
-              {5, _Terms9, Qi4} = restart_test_queue(Qi3),
+              {5, Qi4} = restart_test_queue(Qi3),
               {Qi5, _SeqIdsGuidsF} = queue_index_publish([3,6,8], true, Qi4),
               Qi6 = rabbit_queue_index:deliver([2,3,5,6], Qi5),
               Qi7 = rabbit_queue_index:ack([1,2,3], Qi6),
-              {5, _Terms10, Qi8} = restart_test_queue(Qi7),
+              {5, Qi8} = restart_test_queue(Qi7),
               Qi8
       end),
 
@@ -2020,3 +2048,56 @@ test_queue_recover() ->
               rabbit_amqqueue:internal_delete(QName)
       end),
     passed.
+
+test_configurable_server_properties() ->
+    %% List of the names of the built-in properties do we expect to find
+    BuiltInPropNames = [<<"product">>, <<"version">>, <<"platform">>,
+                        <<"copyright">>, <<"information">>],
+
+    %% Verify that the built-in properties are initially present
+    ActualPropNames = [Key ||
+                         {Key, longstr, _} <- rabbit_reader:server_properties()],
+    true = lists:all(fun (X) -> lists:member(X, ActualPropNames) end,
+                     BuiltInPropNames),
+
+    %% Get the initial server properties configured in the environment
+    {ok, ServerProperties} = application:get_env(rabbit, server_properties),
+
+    %% Helper functions
+    ConsProp = fun (X) -> application:set_env(rabbit,
+                                              server_properties,
+                                              [X | ServerProperties]) end,
+    IsPropPresent = fun (X) -> lists:member(X,
+                                            rabbit_reader:server_properties())
+                    end,
+
+    %% Add a wholly new property of the simplified {KeyAtom, StringValue} form
+    NewSimplifiedProperty = {NewHareKey, NewHareVal} = {hare, "soup"},
+    ConsProp(NewSimplifiedProperty),
+    %% Do we find hare soup, appropriately formatted in the generated properties?
+    ExpectedHareImage = {list_to_binary(atom_to_list(NewHareKey)),
+                         longstr,
+                         list_to_binary(NewHareVal)},
+    true = IsPropPresent(ExpectedHareImage),
+
+    %% Add a wholly new property of the {BinaryKey, Type, Value} form
+    %% and check for it
+    NewProperty = {<<"new-bin-key">>, signedint, -1},
+    ConsProp(NewProperty),
+    %% Do we find the new property?
+    true = IsPropPresent(NewProperty),
+
+    %% Add a property that clobbers a built-in, and verify correct clobbering
+    {NewVerKey, NewVerVal} = NewVersion = {version, "X.Y.Z."},
+    {BinNewVerKey, BinNewVerVal} = {list_to_binary(atom_to_list(NewVerKey)),
+                                    list_to_binary(NewVerVal)},
+    ConsProp(NewVersion),
+    ClobberedServerProps = rabbit_reader:server_properties(),
+    %% Is the clobbering insert present?
+    true = IsPropPresent({BinNewVerKey, longstr, BinNewVerVal}),
+    %% Is the clobbering insert the only thing with the clobbering key?
+    [{BinNewVerKey, longstr, BinNewVerVal}] =
+        [E || {K, longstr, _V} = E <- ClobberedServerProps, K =:= BinNewVerKey],
+
+    application:set_env(rabbit, server_properties, ServerProperties),
+    passed.
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index a1c442d3..3f639423 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -370,16 +370,17 @@ stop_msg_store() ->
     ok = rabbit_sup:stop_child(?PERSISTENT_MSG_STORE),
     ok = rabbit_sup:stop_child(?TRANSIENT_MSG_STORE).
 
-init(QueueName, IsDurable, Recover) ->
-    {DeltaCount, Terms, IndexState} =
-        rabbit_queue_index:init(
-          QueueName, Recover,
-          rabbit_msg_store:successfully_recovered_state(?PERSISTENT_MSG_STORE),
-          fun (Guid) ->
-                  rabbit_msg_store:contains(?PERSISTENT_MSG_STORE, Guid)
-          end),
-    {LowSeqId, NextSeqId, IndexState1} = rabbit_queue_index:bounds(IndexState),
+init(QueueName, IsDurable, false) ->
+    IndexState = rabbit_queue_index:init(QueueName),
+    init(IsDurable, IndexState, 0, [],
+         case IsDurable of
+             true  -> msg_store_client_init(?PERSISTENT_MSG_STORE);
+             false -> undefined
+         end,
+         msg_store_client_init(?TRANSIENT_MSG_STORE));
 
+init(QueueName, true, true) ->
+    Terms = rabbit_queue_index:shutdown_terms(QueueName),
     {PRef, TRef, Terms1} =
         case [persistent_ref, transient_ref] -- proplists:get_keys(Terms) of
             [] -> {proplists:get_value(persistent_ref, Terms),
@@ -387,64 +388,32 @@ init(QueueName, IsDurable, Recover) ->
                    Terms};
             _  -> {rabbit_guid:guid(), rabbit_guid:guid(), []}
         end,
-    DeltaCount1 = proplists:get_value(persistent_count, Terms1, DeltaCount),
-    Delta = case DeltaCount1 == 0 andalso DeltaCount /= undefined of
-                true  -> ?BLANK_DELTA;
-                false -> #delta { start_seq_id = LowSeqId,
-                                  count        = DeltaCount1,
-                                  end_seq_id   = NextSeqId }
-            end,
-    Now = now(),
-    PersistentClient =
-        case IsDurable of
-            true  -> rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, PRef);
-            false -> undefined
-        end,
-    TransientClient  = rabbit_msg_store:client_init(?TRANSIENT_MSG_STORE, TRef),
-    State = #vqstate {
-      q1                   = queue:new(),
-      q2                   = bpqueue:new(),
-      delta                = Delta,
-      q3                   = bpqueue:new(),
-      q4                   = queue:new(),
-      next_seq_id          = NextSeqId,
-      pending_ack          = dict:new(),
-      index_state          = IndexState1,
-      msg_store_clients    = {{PersistentClient, PRef},
-                              {TransientClient, TRef}},
-      on_sync              = ?BLANK_SYNC,
-      durable              = IsDurable,
-      transient_threshold  = NextSeqId,
-
-      len                  = DeltaCount1,
-      persistent_count     = DeltaCount1,
-
-      duration_target      = infinity,
-      target_ram_msg_count = infinity,
-      ram_msg_count        = 0,
-      ram_msg_count_prev   = 0,
-      ram_index_count      = 0,
-      out_counter          = 0,
-      in_counter           = 0,
-      rates                = #rates { egress      = {Now, 0},
-                                      ingress     = {Now, DeltaCount1},
-                                      avg_egress  = 0.0,
-                                      avg_ingress = 0.0,
-                                      timestamp   = Now } },
-    a(maybe_deltas_to_betas(State)).
+    PersistentClient = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE,
+                                                    PRef),
+    TransientClient  = rabbit_msg_store:client_init(?TRANSIENT_MSG_STORE,
+                                                    TRef),
+    {DeltaCount, IndexState} =
+        rabbit_queue_index:recover(
+          QueueName, Terms1,
+          rabbit_msg_store:successfully_recovered_state(?PERSISTENT_MSG_STORE),
+          fun (Guid) ->
+                  rabbit_msg_store:contains(Guid, PersistentClient)
+          end),
+    init(true, IndexState, DeltaCount, Terms1,
+         PersistentClient, TransientClient).
 
 terminate(State) ->
     State1 = #vqstate { persistent_count  = PCount,
                         index_state       = IndexState,
-                        msg_store_clients = {{MSCStateP, PRef},
-                                             {MSCStateT, TRef}} } =
+                        msg_store_clients = {MSCStateP, MSCStateT} } =
         remove_pending_ack(true, tx_commit_index(State)),
-    case MSCStateP of
-        undefined -> ok;
-        _         -> rabbit_msg_store:client_terminate(
-                       MSCStateP, ?PERSISTENT_MSG_STORE)
-    end,
-    rabbit_msg_store:client_terminate(MSCStateT, ?TRANSIENT_MSG_STORE),
+    PRef = case MSCStateP of
+               undefined -> undefined;
+               _         -> ok = rabbit_msg_store:client_terminate(MSCStateP),
+                            rabbit_msg_store:client_ref(MSCStateP)
+           end,
+    ok = rabbit_msg_store:client_terminate(MSCStateT),
+    TRef = rabbit_msg_store:client_ref(MSCStateT),
     Terms = [{persistent_ref, PRef},
              {transient_ref, TRef},
              {persistent_count, PCount}],
@@ -460,44 +429,44 @@ delete_and_terminate(State) ->
     %% deleting it.
     {_PurgeCount, State1} = purge(State),
     State2 = #vqstate { index_state         = IndexState,
-                        msg_store_clients   = {{MSCStateP, PRef},
-                                               {MSCStateT, TRef}} } =
+                        msg_store_clients   = {MSCStateP, MSCStateT} } =
         remove_pending_ack(false, State1),
     IndexState1 = rabbit_queue_index:delete_and_terminate(IndexState),
     case MSCStateP of
         undefined -> ok;
-        _         -> rabbit_msg_store:client_delete_and_terminate(
-                       MSCStateP, ?PERSISTENT_MSG_STORE, PRef)
+        _         -> rabbit_msg_store:client_delete_and_terminate(MSCStateP)
     end,
-    rabbit_msg_store:client_delete_and_terminate(
-      MSCStateT, ?TRANSIENT_MSG_STORE, TRef),
+    rabbit_msg_store:client_delete_and_terminate(MSCStateT),
     a(State2 #vqstate { index_state       = IndexState1,
                         msg_store_clients = undefined }).
 
-purge(State = #vqstate { q4               = Q4,
-                         index_state      = IndexState,
-                         len              = Len,
-                         persistent_count = PCount }) ->
+purge(State = #vqstate { q4                = Q4,
+                         index_state       = IndexState,
+                         msg_store_clients = MSCState,
+                         len               = Len,
+                         persistent_count  = PCount }) ->
     %% TODO: when there are no pending acks, which is a common case,
     %% we could simply wipe the qi instead of issuing delivers and
     %% acks for all the messages.
     {LensByStore, IndexState1} = remove_queue_entries(
                                    fun rabbit_misc:queue_fold/3, Q4,
-                                   orddict:new(), IndexState),
-    {LensByStore1, State1 = #vqstate { q1 = Q1, index_state = IndexState2 }} =
+                                   orddict:new(), IndexState, MSCState),
+    {LensByStore1, State1 = #vqstate { q1                = Q1,
+                                       index_state       = IndexState2,
+                                       msg_store_clients = MSCState1 }} =
         purge_betas_and_deltas(LensByStore,
                                State #vqstate { q4          = queue:new(),
                                                 index_state = IndexState1 }),
     {LensByStore2, IndexState3} = remove_queue_entries(
                                     fun rabbit_misc:queue_fold/3, Q1,
-                                    LensByStore1, IndexState2),
+                                    LensByStore1, IndexState2, MSCState1),
     PCount1 = PCount - find_persistent_count(LensByStore2),
-    {Len, a(State1 #vqstate { q1               = queue:new(),
-                              index_state      = IndexState3,
-                              len              = 0,
-                              ram_msg_count    = 0,
-                              ram_index_count  = 0,
-                              persistent_count = PCount1 })}.
+    {Len, a(State1 #vqstate { q1                = queue:new(),
+                              index_state       = IndexState3,
+                              len               = 0,
+                              ram_msg_count     = 0,
+                              ram_index_count   = 0,
+                              persistent_count  = PCount1 })}.
 
 publish(Msg, MsgProps, State) ->
     {_SeqId, State1} = publish(Msg, MsgProps, false, false, State),
@@ -574,7 +543,7 @@ read_msg(MsgStatus = #msg_status { msg           = undefined,
          State = #vqstate { ram_msg_count     = RamMsgCount,
                             msg_store_clients = MSCState}) ->
     {{ok, Msg = #basic_message {}}, MSCState1} =
-        read_from_msg_store(MSCState, IsPersistent, Guid),
+        msg_store_read(MSCState, IsPersistent, Guid),
     {MsgStatus #msg_status { msg = Msg },
      State #vqstate { ram_msg_count     = RamMsgCount + 1,
                       msg_store_clients = MSCState1 }};
@@ -589,20 +558,22 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
                               is_delivered  = IsDelivered,
                               msg_on_disk   = MsgOnDisk,
                               index_on_disk = IndexOnDisk },
-               State = #vqstate {ram_msg_count    = RamMsgCount,
-                                 out_counter      = OutCount,
-                                 index_state      = IndexState,
-                                 len              = Len,
-                                 persistent_count = PCount,
-                                 pending_ack      = PA }) ->
+               State = #vqstate {ram_msg_count     = RamMsgCount,
+                                 out_counter       = OutCount,
+                                 index_state       = IndexState,
+                                 msg_store_clients = MSCState,
+                                 len               = Len,
+                                 persistent_count  = PCount,
+                                 pending_ack       = PA }) ->
     %% 1. Mark it delivered if necessary
     IndexState1 = maybe_write_delivered(
                     IndexOnDisk andalso not IsDelivered,
                     SeqId, IndexState),
 
     %% 2. Remove from msg_store and queue index, if necessary
-    MsgStore = find_msg_store(IsPersistent),
-    Rem = fun () -> ok = rabbit_msg_store:remove(MsgStore, [Guid]) end,
+    Rem = fun () ->
+                  ok = msg_store_remove(MSCState, IsPersistent, [Guid])
+          end,
     Ack = fun () -> rabbit_queue_index:ack([SeqId], IndexState1) end,
     IndexState2 =
         case {AckRequired, MsgOnDisk, IndexOnDisk, IsPersistent} of
@@ -634,7 +605,7 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
                         pending_ack      = PA1 })}.
 
 ack(AckTags, State) ->
-    a(ack(fun rabbit_msg_store:remove/2,
+    a(ack(fun msg_store_remove/3,
           fun (_AckEntry, State1) -> State1 end,
           AckTags, State)).
 
@@ -643,30 +614,32 @@ tx_publish(Txn, Msg = #basic_message { is_persistent = IsPersistent }, MsgProps,
                               msg_store_clients = MSCState }) ->
     Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn),
     store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProps} | Pubs] }),
-    a(case IsPersistent andalso IsDurable of
-          true  -> MsgStatus = msg_status(true, undefined, Msg, MsgProps),
-                   {#msg_status { msg_on_disk = true }, MSCState1} =
-                       maybe_write_msg_to_disk(false, MsgStatus, MSCState),
-                   State #vqstate { msg_store_clients = MSCState1 };
-          false -> State
-      end).
+    case IsPersistent andalso IsDurable of
+        true  -> MsgStatus = msg_status(true, undefined, Msg, MsgProps),
+                 #msg_status { msg_on_disk = true } =
+                     maybe_write_msg_to_disk(false, MsgStatus, MSCState);
+        false -> ok
+    end,
+    a(State).
 
 tx_ack(Txn, AckTags, State) ->
     Tx = #tx { pending_acks = Acks } = lookup_tx(Txn),
     store_tx(Txn, Tx #tx { pending_acks = [AckTags | Acks] }),
     State.
 
-tx_rollback(Txn, State = #vqstate { durable = IsDurable }) ->
+tx_rollback(Txn, State = #vqstate { durable           = IsDurable,
+                                    msg_store_clients = MSCState }) ->
     #tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn),
     erase_tx(Txn),
     ok = case IsDurable of
-             true  -> rabbit_msg_store:remove(?PERSISTENT_MSG_STORE,
-                                              persistent_guids(Pubs));
+             true  -> msg_store_remove(MSCState, true, persistent_guids(Pubs));
              false -> ok
          end,
     {lists:append(AckTags), a(State)}.
 
-tx_commit(Txn, Fun, MsgPropsFun, State = #vqstate { durable = IsDurable }) ->
+tx_commit(Txn, Fun, MsgPropsFun,
+          State = #vqstate { durable           = IsDurable,
+                             msg_store_clients = MSCState }) ->
     #tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn),
     erase_tx(Txn),
     AckTags1 = lists:append(AckTags),
@@ -674,8 +647,8 @@ tx_commit(Txn, Fun, MsgPropsFun, State = #vqstate { durable = IsDurable }) ->
     HasPersistentPubs = PersistentGuids =/= [],
     {AckTags1,
      a(case IsDurable andalso HasPersistentPubs of
-           true  -> ok = rabbit_msg_store:sync(
-                           ?PERSISTENT_MSG_STORE, PersistentGuids,
+           true  -> ok = msg_store_sync(
+                           MSCState, true, PersistentGuids,
                            msg_store_callback(PersistentGuids, Pubs, AckTags1,
                                               Fun, MsgPropsFun)),
                     State;
@@ -685,7 +658,7 @@ tx_commit(Txn, Fun, MsgPropsFun, State = #vqstate { durable = IsDurable }) ->
 
 requeue(AckTags, MsgPropsFun, State) ->
     a(reduce_memory_use(
-        ack(fun rabbit_msg_store:release/2,
+        ack(fun msg_store_release/3,
             fun (#msg_status { msg = Msg, msg_props = MsgProps }, State1) ->
                     {_SeqId, State2} = publish(Msg, MsgPropsFun(MsgProps),
                                                true, false, State1),
@@ -693,7 +666,7 @@ requeue(AckTags, MsgPropsFun, State) ->
                 ({IsPersistent, Guid, MsgProps}, State1) ->
                     #vqstate { msg_store_clients = MSCState } = State1,
                     {{ok, Msg = #basic_message{}}, MSCState1} =
-                        read_from_msg_store(MSCState, IsPersistent, Guid),
+                        msg_store_read(MSCState, IsPersistent, Guid),
                     State2 = State1 #vqstate { msg_store_clients = MSCState1 },
                     {_SeqId, State3} = publish(Msg, MsgPropsFun(MsgProps),
                                                true, true, State2),
@@ -851,22 +824,47 @@ msg_status(IsPersistent, SeqId, Msg = #basic_message { guid = Guid },
                   msg_on_disk = false, index_on_disk = false,
                   msg_props = MsgProps }.
 
-find_msg_store(true)  -> ?PERSISTENT_MSG_STORE;
-find_msg_store(false) -> ?TRANSIENT_MSG_STORE.
-
-with_msg_store_state({{MSCStateP, PRef}, MSCStateT}, true, Fun) ->
-    {Result, MSCStateP1} = Fun(?PERSISTENT_MSG_STORE, MSCStateP),
-    {Result, {{MSCStateP1, PRef}, MSCStateT}};
-with_msg_store_state({MSCStateP, {MSCStateT, TRef}}, false, Fun) ->
-    {Result, MSCStateT1} = Fun(?TRANSIENT_MSG_STORE, MSCStateT),
-    {Result, {MSCStateP, {MSCStateT1, TRef}}}.
+with_msg_store_state({MSCStateP, MSCStateT},  true, Fun) ->
+    {Result, MSCStateP1} = Fun(MSCStateP),
+    {Result, {MSCStateP1, MSCStateT}};
+with_msg_store_state({MSCStateP, MSCStateT}, false, Fun) ->
+    {Result, MSCStateT1} = Fun(MSCStateT),
+    {Result, {MSCStateP, MSCStateT1}}.
+
+with_immutable_msg_store_state(MSCState, IsPersistent, Fun) ->
+    {Res, MSCState} = with_msg_store_state(MSCState, IsPersistent,
+                                           fun (MSCState1) ->
+                                                   {Fun(MSCState1), MSCState1}
+                                           end),
+    Res.
+
+msg_store_client_init(MsgStore) ->
+    rabbit_msg_store:client_init(MsgStore, rabbit_guid:guid()).
+
+msg_store_write(MSCState, IsPersistent, Guid, Msg) ->
+    with_immutable_msg_store_state(
+      MSCState, IsPersistent,
+      fun (MSCState1) -> rabbit_msg_store:write(Guid, Msg, MSCState1) end).
 
-read_from_msg_store(MSCState, IsPersistent, Guid) ->
+msg_store_read(MSCState, IsPersistent, Guid) ->
     with_msg_store_state(
       MSCState, IsPersistent,
-      fun (MsgStore, MSCState1) ->
-              rabbit_msg_store:read(MsgStore, Guid, MSCState1)
-      end).
+      fun (MSCState1) -> rabbit_msg_store:read(Guid, MSCState1) end).
+
+msg_store_remove(MSCState, IsPersistent, Guids) ->
+    with_immutable_msg_store_state(
+      MSCState, IsPersistent,
+      fun (MCSState1) -> rabbit_msg_store:remove(Guids, MCSState1) end).
+
+msg_store_release(MSCState, IsPersistent, Guids) ->
+    with_immutable_msg_store_state(
+      MSCState, IsPersistent,
+      fun (MCSState1) -> rabbit_msg_store:release(Guids, MCSState1) end).
+
+msg_store_sync(MSCState, IsPersistent, Guids, Callback) ->
+    with_immutable_msg_store_state(
+      MSCState, IsPersistent,
+      fun (MSCState1) -> rabbit_msg_store:sync(Guids, Callback, MSCState1) end).
 
 maybe_write_delivered(false, _SeqId, IndexState) ->
     IndexState;
@@ -950,6 +948,49 @@ update_rate(Now, Then, Count, {OThen, OCount}) ->
 %% Internal major helpers for Public API
 %%----------------------------------------------------------------------------
 
+init(IsDurable, IndexState, DeltaCount, Terms,
+     PersistentClient, TransientClient) ->
+    {LowSeqId, NextSeqId, IndexState1} = rabbit_queue_index:bounds(IndexState),
+
+    DeltaCount1 = proplists:get_value(persistent_count, Terms, DeltaCount),
+    Delta = case DeltaCount1 == 0 andalso DeltaCount /= undefined of
+                true  -> ?BLANK_DELTA;
+                false -> #delta { start_seq_id = LowSeqId,
+                                  count        = DeltaCount1,
+                                  end_seq_id   = NextSeqId }
+            end,
+    Now = now(),
+    State = #vqstate {
+      q1                   = queue:new(),
+      q2                   = bpqueue:new(),
+      delta                = Delta,
+      q3                   = bpqueue:new(),
+      q4                   = queue:new(),
+      next_seq_id          = NextSeqId,
+      pending_ack          = dict:new(),
+      index_state          = IndexState1,
+      msg_store_clients    = {PersistentClient, TransientClient},
+      on_sync              = ?BLANK_SYNC,
+      durable              = IsDurable,
+      transient_threshold  = NextSeqId,
+
+      len                  = DeltaCount1,
+      persistent_count     = DeltaCount1,
+
+      duration_target      = infinity,
+      target_ram_msg_count = infinity,
+      ram_msg_count        = 0,
+      ram_msg_count_prev   = 0,
+      ram_index_count      = 0,
+      out_counter          = 0,
+      in_counter           = 0,
+      rates                = #rates { egress      = {Now, 0},
+                                      ingress     = {Now, DeltaCount1},
+                                      avg_egress  = 0.0,
+                                      avg_ingress = 0.0,
+                                      timestamp   = Now } },
+    a(maybe_deltas_to_betas(State)).
+
 msg_store_callback(PersistentGuids, Pubs, AckTags, Fun, MsgPropsFun) ->
     Self = self(),
     F = fun () -> rabbit_amqqueue:maybe_run_queue_via_backing_queue(
@@ -959,13 +1000,17 @@ msg_store_callback(PersistentGuids, Pubs, AckTags, Fun, MsgPropsFun) ->
                           end)
         end,
     fun () -> spawn(fun () -> ok = rabbit_misc:with_exit_handler(
-                                     fun () -> rabbit_msg_store:remove(
-                                                 ?PERSISTENT_MSG_STORE,
+                                     fun () -> remove_persistent_messages(
                                                  PersistentGuids)
                                      end, F)
                     end)
     end.
 
+remove_persistent_messages(Guids) ->
+    PersistentClient = msg_store_client_init(?PERSISTENT_MSG_STORE),
+    ok = rabbit_msg_store:remove(Guids, PersistentClient),
+    rabbit_msg_store:client_delete_and_terminate(PersistentClient).
+
 tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun, MsgPropsFun,
                          State = #vqstate {
                            on_sync     = OnSync = #sync {
@@ -1031,13 +1076,14 @@ tx_commit_index(State = #vqstate { on_sync = #sync {
       State1 #vqstate { index_state = IndexState1, on_sync = ?BLANK_SYNC }).
 
 purge_betas_and_deltas(LensByStore,
-                       State = #vqstate { q3          = Q3,
-                                          index_state = IndexState }) ->
+                       State = #vqstate { q3                = Q3,
+                                          index_state       = IndexState,
+                                          msg_store_clients = MSCState }) ->
     case bpqueue:is_empty(Q3) of
         true  -> {LensByStore, State};
-        false -> {LensByStore1, IndexState1} = remove_queue_entries(
-                                                 fun beta_fold/3, Q3,
-                                                 LensByStore, IndexState),
+        false -> {LensByStore1, IndexState1} =
+                     remove_queue_entries(fun beta_fold/3, Q3,
+                                          LensByStore, IndexState, MSCState),
                  purge_betas_and_deltas(LensByStore1,
                                         maybe_deltas_to_betas(
                                           State #vqstate {
@@ -1045,11 +1091,11 @@ purge_betas_and_deltas(LensByStore,
                                             index_state = IndexState1 }))
     end.
 
-remove_queue_entries(Fold, Q, LensByStore, IndexState) ->
+remove_queue_entries(Fold, Q, LensByStore, IndexState, MSCState) ->
     {GuidsByStore, Delivers, Acks} =
         Fold(fun remove_queue_entries1/2, {orddict:new(), [], []}, Q),
-    ok = orddict:fold(fun (MsgStore, Guids, ok) ->
-                              rabbit_msg_store:remove(MsgStore, Guids)
+    ok = orddict:fold(fun (IsPersistent, Guids, ok) ->
+                              msg_store_remove(MSCState, IsPersistent, Guids)
                       end, ok, GuidsByStore),
     {sum_guids_by_store_to_len(LensByStore, GuidsByStore),
      rabbit_queue_index:ack(Acks,
@@ -1061,8 +1107,7 @@ remove_queue_entries1(
                 index_on_disk = IndexOnDisk, is_persistent = IsPersistent },
   {GuidsByStore, Delivers, Acks}) ->
     {case MsgOnDisk of
-         true  -> rabbit_misc:orddict_cons(find_msg_store(IsPersistent), Guid,
-                                           GuidsByStore);
+         true  -> rabbit_misc:orddict_cons(IsPersistent, Guid, GuidsByStore);
          false -> GuidsByStore
      end,
      cons_if(IndexOnDisk andalso not IsDelivered, SeqId, Delivers),
@@ -1070,8 +1115,8 @@ remove_queue_entries1(
 
 sum_guids_by_store_to_len(LensByStore, GuidsByStore) ->
     orddict:fold(
-      fun (MsgStore, Guids, LensByStore1) ->
-              orddict:update_counter(MsgStore, length(Guids), LensByStore1)
+      fun (IsPersistent, Guids, LensByStore1) ->
+              orddict:update_counter(IsPersistent, length(Guids), LensByStore1)
       end, LensByStore, GuidsByStore).
 
 %%----------------------------------------------------------------------------
@@ -1103,8 +1148,8 @@ publish(Msg = #basic_message { is_persistent = IsPersistent },
                               ram_msg_count    = RamMsgCount + 1}}.
 
 maybe_write_msg_to_disk(_Force, MsgStatus = #msg_status {
-                                  msg_on_disk = true }, MSCState) ->
-    {MsgStatus, MSCState};
+                                  msg_on_disk = true }, _MSCState) ->
+    MsgStatus;
 maybe_write_msg_to_disk(Force, MsgStatus = #msg_status {
                                  msg = Msg, guid = Guid,
                                  is_persistent = IsPersistent }, MSCState)
@@ -1113,15 +1158,10 @@ maybe_write_msg_to_disk(Force, MsgStatus = #msg_status {
              %% don't persist any recoverable decoded properties
              content = rabbit_binary_parser:clear_decoded_content(
                          Msg #basic_message.content)},
-    {ok, MSCState1} = with_msg_store_state(
-                        MSCState, IsPersistent,
-                        fun (MsgStore, MSCState2) ->
-                                rabbit_msg_store:write(MsgStore, Guid, Msg1,
-                                                       MSCState2)
-                        end),
-    {MsgStatus #msg_status { msg_on_disk = true }, MSCState1};
-maybe_write_msg_to_disk(_Force, MsgStatus, MSCState) ->
-    {MsgStatus, MSCState}.
+    ok = msg_store_write(MSCState, IsPersistent, Guid, Msg1),
+    MsgStatus #msg_status { msg_on_disk = true };
+maybe_write_msg_to_disk(_Force, MsgStatus, _MSCState) ->
+    MsgStatus.
 
 maybe_write_index_to_disk(_Force, MsgStatus = #msg_status {
                                     index_on_disk = true }, IndexState) ->
@@ -1145,12 +1185,10 @@ maybe_write_index_to_disk(_Force, MsgStatus, IndexState) ->
 maybe_write_to_disk(ForceMsg, ForceIndex, MsgStatus,
                     State = #vqstate { index_state       = IndexState,
                                        msg_store_clients = MSCState }) ->
-    {MsgStatus1, MSCState1}   = maybe_write_msg_to_disk(
-                                  ForceMsg, MsgStatus, MSCState),
-    {MsgStatus2, IndexState1} = maybe_write_index_to_disk(
-                                  ForceIndex, MsgStatus1, IndexState),
-    {MsgStatus2, State #vqstate { index_state       = IndexState1,
-                                  msg_store_clients = MSCState1 }}.
+    MsgStatus1 = maybe_write_msg_to_disk(ForceMsg, MsgStatus, MSCState),
+    {MsgStatus2, IndexState1} =
+        maybe_write_index_to_disk(ForceIndex, MsgStatus1, IndexState),
+    {MsgStatus2, State #vqstate { index_state = IndexState1 }}.
 
 %%----------------------------------------------------------------------------
 %% Internal gubbins for acks
@@ -1168,22 +1206,23 @@ record_pending_ack(#msg_status { seq_id        = SeqId,
     dict:store(SeqId, AckEntry, PA).
 
 remove_pending_ack(KeepPersistent,
-                   State = #vqstate { pending_ack = PA,
-                                      index_state = IndexState }) ->
+                   State = #vqstate { pending_ack       = PA,
+                                      index_state       = IndexState,
+                                      msg_store_clients = MSCState }) ->
     {SeqIds, GuidsByStore} = dict:fold(fun accumulate_ack/3,
                                        {[], orddict:new()}, PA),
     State1 = State #vqstate { pending_ack = dict:new() },
     case KeepPersistent of
-        true  -> case orddict:find(?TRANSIENT_MSG_STORE, GuidsByStore) of
+        true  -> case orddict:find(false, GuidsByStore) of
                      error       -> State1;
-                     {ok, Guids} -> ok = rabbit_msg_store:remove(
-                                           ?TRANSIENT_MSG_STORE, Guids),
+                     {ok, Guids} -> ok = msg_store_remove(MSCState, false,
+                                                          Guids),
                                     State1
                  end;
         false -> IndexState1 = rabbit_queue_index:ack(SeqIds, IndexState),
                  ok = orddict:fold(
-                        fun (MsgStore, Guids, ok) ->
-                                rabbit_msg_store:remove(MsgStore, Guids)
+                        fun (IsPersistent, Guids, ok) ->
+                                msg_store_remove(MSCState, IsPersistent, Guids)
                         end, ok, GuidsByStore),
                  State1 #vqstate { index_state = IndexState1 }
     end.
@@ -1191,8 +1230,10 @@ remove_pending_ack(KeepPersistent,
 ack(_MsgStoreFun, _Fun, [], State) ->
     State;
 ack(MsgStoreFun, Fun, AckTags, State) ->
-    {{SeqIds, GuidsByStore}, State1 = #vqstate { index_state      = IndexState,
-                                                 persistent_count = PCount }} =
+    {{SeqIds, GuidsByStore},
+     State1 = #vqstate { index_state       = IndexState,
+                         msg_store_clients = MSCState,
+                         persistent_count  = PCount }} =
         lists:foldl(
           fun (SeqId, {Acc, State2 = #vqstate { pending_ack = PA }}) ->
                   {ok, AckEntry} = dict:find(SeqId, PA),
@@ -1201,8 +1242,8 @@ ack(MsgStoreFun, Fun, AckTags, State) ->
                                    pending_ack = dict:erase(SeqId, PA) })}
           end, {{[], orddict:new()}, State}, AckTags),
     IndexState1 = rabbit_queue_index:ack(SeqIds, IndexState),
-    ok = orddict:fold(fun (MsgStore, Guids, ok) ->
-                              MsgStoreFun(MsgStore, Guids)
+    ok = orddict:fold(fun (IsPersistent, Guids, ok) ->
+                              MsgStoreFun(MSCState, IsPersistent, Guids)
                       end, ok, GuidsByStore),
     PCount1 = PCount - find_persistent_count(sum_guids_by_store_to_len(
                                                orddict:new(), GuidsByStore)),
@@ -1215,10 +1256,10 @@ accumulate_ack(_SeqId, #msg_status { is_persistent = false, %% ASSERTIONS
     Acc;
 accumulate_ack(SeqId, {IsPersistent, Guid, _MsgProps}, {SeqIdsAcc, Dict}) ->
     {cons_if(IsPersistent, SeqId, SeqIdsAcc),
-     rabbit_misc:orddict_cons(find_msg_store(IsPersistent), Guid, Dict)}.
+     rabbit_misc:orddict_cons(IsPersistent, Guid, Dict)}.
 
 find_persistent_count(LensByStore) ->
-    case orddict:find(?PERSISTENT_MSG_STORE, LensByStore) of
+    case orddict:find(true, LensByStore) of
         error     -> 0;
         {ok, Len} -> Len
     end.
@@ -1360,46 +1401,40 @@ maybe_deltas_to_betas(State = #vqstate {
                         delta                = Delta,
                         q3                   = Q3,
                         index_state          = IndexState,
-                        target_ram_msg_count = TargetRamMsgCount,
                         transient_threshold  = TransientThreshold }) ->
-    case bpqueue:is_empty(Q3) orelse (TargetRamMsgCount /= 0) of
-        false ->
-            State;
-        true ->
-            #delta { start_seq_id = DeltaSeqId,
-                     count        = DeltaCount,
-                     end_seq_id   = DeltaSeqIdEnd } = Delta,
-            DeltaSeqId1 =
-                lists:min([rabbit_queue_index:next_segment_boundary(DeltaSeqId),
-                           DeltaSeqIdEnd]),
-            {List, IndexState1} =
-                rabbit_queue_index:read(DeltaSeqId, DeltaSeqId1, IndexState),
-            {Q3a, IndexState2} = betas_from_index_entries(
-                                   List, TransientThreshold, IndexState1),
-            State1 = State #vqstate { index_state = IndexState2 },
-            case bpqueue:len(Q3a) of
+    #delta { start_seq_id = DeltaSeqId,
+             count        = DeltaCount,
+             end_seq_id   = DeltaSeqIdEnd } = Delta,
+    DeltaSeqId1 =
+        lists:min([rabbit_queue_index:next_segment_boundary(DeltaSeqId),
+                   DeltaSeqIdEnd]),
+    {List, IndexState1} =
+        rabbit_queue_index:read(DeltaSeqId, DeltaSeqId1, IndexState),
+    {Q3a, IndexState2} =
+        betas_from_index_entries(List, TransientThreshold, IndexState1),
+    State1 = State #vqstate { index_state = IndexState2 },
+    case bpqueue:len(Q3a) of
+        0 ->
+            %% we ignored every message in the segment due to it being
+            %% transient and below the threshold
+            maybe_deltas_to_betas(
+              State1 #vqstate {
+                delta = Delta #delta { start_seq_id = DeltaSeqId1 }});
+        Q3aLen ->
+            Q3b = bpqueue:join(Q3, Q3a),
+            case DeltaCount - Q3aLen of
                 0 ->
-                    %% we ignored every message in the segment due to
-                    %% it being transient and below the threshold
-                    maybe_deltas_to_betas(
-                      State #vqstate {
-                        delta = Delta #delta { start_seq_id = DeltaSeqId1 }});
-                Q3aLen ->
-                    Q3b = bpqueue:join(Q3, Q3a),
-                    case DeltaCount - Q3aLen of
-                        0 ->
-                            %% delta is now empty, but it wasn't
-                            %% before, so can now join q2 onto q3
-                            State1 #vqstate { q2    = bpqueue:new(),
-                                              delta = ?BLANK_DELTA,
-                                              q3    = bpqueue:join(Q3b, Q2) };
-                        N when N > 0 ->
-                            Delta1 = #delta { start_seq_id = DeltaSeqId1,
-                                              count        = N,
-                                              end_seq_id   = DeltaSeqIdEnd },
-                            State1 #vqstate { delta = Delta1,
-                                              q3    = Q3b }
-                    end
+                    %% delta is now empty, but it wasn't before, so
+                    %% can now join q2 onto q3
+                    State1 #vqstate { q2    = bpqueue:new(),
+                                      delta = ?BLANK_DELTA,
+                                      q3    = bpqueue:join(Q3b, Q2) };
+                N when N > 0 ->
+                    Delta1 = #delta { start_seq_id = DeltaSeqId1,
+                                      count        = N,
+                                      end_seq_id   = DeltaSeqIdEnd },
+                    State1 #vqstate { delta = Delta1,
+                                      q3    = Q3b }
             end
     end.