Merge from default.

41 files changed, 1589 insertions, 1676 deletions

diff --git a/Makefile b/Makefile
index d3f052f6..e0d5744c 100644
--- a/Makefile
+++ b/Makefile
@@ -178,9 +178,6 @@ start-rabbit-on-node: all
 stop-rabbit-on-node: all
 	echo "rabbit:stop()." | $(ERL_CALL)
 
-force-snapshot: all
-	echo "rabbit_persister:force_snapshot()." | $(ERL_CALL)
-
 set-memory-alarm: all
 	echo "alarm_handler:set_alarm({vm_memory_high_watermark, []})." | \
 	$(ERL_CALL)
diff --git a/docs/html-to-website-xml.xsl b/docs/html-to-website-xml.xsl
index ec8f87e5..4bfcf6ca 100644
--- a/docs/html-to-website-xml.xsl
+++ b/docs/html-to-website-xml.xsl
@@ -1,6 +1,7 @@
 <?xml version='1.0'?>
 <xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform"
                 xmlns:doc="http://www.rabbitmq.com/namespaces/ad-hoc/doc"
+                xmlns="http://www.w3.org/1999/xhtml"
                 version='1.0'>
 
 <xsl:param name="original"/>
@@ -10,10 +11,16 @@
 <xsl:template match="*"/>
 
 <!-- Copy every element through -->
-<xsl:template match="@*|node()">
-  <xsl:copy><xsl:apply-templates select="@*|node()"/></xsl:copy>
+<xsl:template match="*">
+  <xsl:element name="{name()}" namespace="http://www.w3.org/1999/xhtml">
+    <xsl:apply-templates select="@*|node()"/>
+  </xsl:element>
 </xsl:template>
 
+<xsl:template match="@*">
+  <xsl:copy/>
+</xsl:template>
+  
 <!-- Copy the root node, and munge the outer part of the page -->
 <xsl:template match="/html">
 <xsl:processing-instruction name="xml-stylesheet">type="text/xml" href="page.xsl"</xsl:processing-instruction>
diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index 17d05a99..6c33ef8b 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -6,7 +6,6 @@
   {registered, [rabbit_amqqueue_sup,
                 rabbit_log,
                 rabbit_node_monitor,
-                rabbit_persister,
                 rabbit_router,
                 rabbit_sup,
                 rabbit_tcp_client_sup]},
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 2b4347dd..8c8e12a1 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -69,12 +69,13 @@
                         is_persistent}).
 
 -record(ssl_socket, {tcp, ssl}).
--record(delivery, {mandatory, immediate, txn, sender, message}).
+-record(delivery, {mandatory, immediate, txn, sender, message,
+                   msg_seq_no}).
 -record(amqp_error, {name, explanation = "", method = none}).
 
 -record(event, {type, props, timestamp}).
 
--record(message_properties, {expiry}).
+-record(message_properties, {expiry, needs_confirming = false}).
 
 %%----------------------------------------------------------------------------
 
diff --git a/include/rabbit_backing_queue_spec.hrl b/include/rabbit_backing_queue_spec.hrl
index 20230b24..f67c6f46 100644
--- a/include/rabbit_backing_queue_spec.hrl
+++ b/include/rabbit_backing_queue_spec.hrl
@@ -37,6 +37,7 @@
 -type(attempt_recovery() :: boolean()).
 -type(purged_msg_count() :: non_neg_integer()).
 -type(ack_required() :: boolean()).
+-type(confirm_required() :: boolean()).
 -type(message_properties_transformer() ::
         fun ((rabbit_types:message_properties())
              -> rabbit_types:message_properties())).
@@ -57,7 +58,7 @@
         (fun ((rabbit_types:message_properties()) -> boolean()), state())
         -> state()).
 -spec(fetch/2 :: (ack_required(), state()) -> {fetch_result(), state()}).
--spec(ack/2 :: ([ack()], state()) -> state()).
+-spec(ack/2 :: ([ack()], state()) -> {[rabbit_guid:guid()], state()}).
 -spec(tx_publish/4 :: (rabbit_types:txn(), rabbit_types:basic_message(),
                        rabbit_types:message_properties(), state()) -> state()).
 -spec(tx_ack/3 :: (rabbit_types:txn(), [ack()], state()) -> state()).
diff --git a/packaging/RPMS/Fedora/rabbitmq-server.spec b/packaging/RPMS/Fedora/rabbitmq-server.spec
index 209a90ee..b37f7ab1 100644
--- a/packaging/RPMS/Fedora/rabbitmq-server.spec
+++ b/packaging/RPMS/Fedora/rabbitmq-server.spec
@@ -9,8 +9,7 @@ Source: http://www.rabbitmq.com/releases/rabbitmq-server/v%{version}/%{name}-%{v
 Source1: rabbitmq-server.init
 Source2: rabbitmq-script-wrapper
 Source3: rabbitmq-server.logrotate
-Source4: rabbitmq-asroot-script-wrapper
-Source5: rabbitmq-server.ocf
+Source4: rabbitmq-server.ocf
 URL: http://www.rabbitmq.com/
 BuildArch: noarch
 BuildRequires: erlang >= R12B-3, python-simplejson, xmlto, libxslt
@@ -29,8 +28,7 @@ scalable implementation of an AMQP broker.
 %define _rabbit_libdir %{_exec_prefix}/lib/rabbitmq
 %define _rabbit_erllibdir %{_rabbit_libdir}/lib/rabbitmq_server-%{version}
 %define _rabbit_wrapper %{_builddir}/`basename %{S:2}`
-%define _rabbit_asroot_wrapper %{_builddir}/`basename %{S:4}`
-%define _rabbit_server_ocf %{_builddir}/`basename %{S:5}`
+%define _rabbit_server_ocf %{_builddir}/`basename %{S:4}`
 %define _plugins_state_dir %{_localstatedir}/lib/rabbitmq/plugins
 
 %define _maindir %{buildroot}%{_rabbit_erllibdir}
@@ -40,8 +38,7 @@ scalable implementation of an AMQP broker.
 
 %build
 cp %{S:2} %{_rabbit_wrapper}
-cp %{S:4} %{_rabbit_asroot_wrapper}
-cp %{S:5} %{_rabbit_server_ocf}
+cp %{S:4} %{_rabbit_server_ocf}
 make %{?_smp_mflags}
 
 %install
@@ -127,6 +124,9 @@ done
 rm -rf %{buildroot}
 
 %changelog
+* Mon Nov 29 2010 rob@rabbitmq.com 2.2.0-1
+- New Upstream Release
+
 * Tue Oct 19 2010 vlad@rabbitmq.com 2.1.1-1
 - New Upstream Release
 
diff --git a/packaging/common/rabbitmq-asroot-script-wrapper b/packaging/common/rabbitmq-asroot-script-wrapper
deleted file mode 100644
index 693a6f0b..00000000
--- a/packaging/common/rabbitmq-asroot-script-wrapper
+++ /dev/null
@@ -1,45 +0,0 @@
-#!/bin/sh
-##   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 Developers of the Original Code are LShift Ltd,
-##   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
-##
-##   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
-##   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
-##   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
-##   Technologies LLC, and Rabbit Technologies Ltd.
-##
-##   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
-##   Ltd. Portions created by Cohesive Financial Technologies LLC are
-##   Copyright (C) 2007-2010 Cohesive Financial Technologies
-##   LLC. Portions created by Rabbit Technologies Ltd are Copyright
-##   (C) 2007-2010 Rabbit Technologies Ltd.
-##
-##   All Rights Reserved.
-##
-##   Contributor(s): ______________________________________.
-##
-
-cd /var/lib/rabbitmq
-
-SCRIPT=`basename $0`
-
-if [ `id -u` = 0 ] ; then
-    /usr/lib/rabbitmq/bin/${SCRIPT} "$@"
-else
-    echo
-    echo "Only root should run ${SCRIPT}"
-    echo
-    exit 1
-fi
-
diff --git a/packaging/debs/Debian/debian/changelog b/packaging/debs/Debian/debian/changelog
index e81fda24..a60e691d 100644
--- a/packaging/debs/Debian/debian/changelog
+++ b/packaging/debs/Debian/debian/changelog
@@ -1,3 +1,9 @@
+rabbitmq-server (2.2.0-1) lucid; urgency=low
+
+  * New Upstream Release
+
+ -- Rob Harrop <rob@rabbitmq.com>  Mon, 29 Nov 2010 12:24:48 +0000
+
 rabbitmq-server (2.1.1-1) lucid; urgency=low
 
   * New Upstream Release
diff --git a/packaging/macports/Makefile b/packaging/macports/Makefile
index ee79c95a..47da02dc 100644
--- a/packaging/macports/Makefile
+++ b/packaging/macports/Makefile
@@ -38,9 +38,7 @@ $(DEST)/Portfile: Portfile.in
 # needs vars such as HOME to be set.  So we have to set them
 # explicitly.
 macports: dirs $(DEST)/Portfile
-	for f in rabbitmq-asroot-script-wrapper rabbitmq-script-wrapper ; do \
-	  cp $(COMMON_DIR)/$$f $(DEST)/files ; \
-	done
+	cp $(COMMON_DIR)/rabbitmq-script-wrapper $(DEST)/files
 	sed -i -e 's|@SU_RABBITMQ_SH_C@|SHELL=/bin/sh HOME=/var/lib/rabbitmq USER=rabbitmq LOGNAME=rabbitmq PATH="$$(eval `PATH=MACPORTS_PREFIX/bin /usr/libexec/path_helper -s`; echo $$PATH)" su -m rabbitmq -c|' \
 	    $(DEST)/files/rabbitmq-script-wrapper
 	cp patch-org.macports.rabbitmq-server.plist.diff $(DEST)/files
diff --git a/packaging/macports/Portfile.in b/packaging/macports/Portfile.in
index ce6b1e34..f8417b83 100644
--- a/packaging/macports/Portfile.in
+++ b/packaging/macports/Portfile.in
@@ -7,6 +7,8 @@ version		@VERSION@
 categories	net 
 maintainers	paperplanes.de:meyer rabbitmq.com:tonyg openmaintainer
 platforms	darwin
+supported_archs	noarch
+
 description	The RabbitMQ AMQP Server
 long_description	\
     RabbitMQ is an implementation of AMQP, the emerging standard for \
@@ -31,17 +33,13 @@ checksums \
 depends_lib	port:erlang
 depends_build	port:libxslt
 
-platform darwin 7 {
-    depends_build-append port:py25-simplejson
-    build.args      PYTHON=${prefix}/bin/python2.5
-}
 platform darwin 8 {
-    depends_build-append port:py25-simplejson
-    build.args      PYTHON=${prefix}/bin/python2.5
+    depends_build-append port:py26-simplejson
+    build.args      PYTHON=${prefix}/bin/python2.6
 }
 platform darwin 9 {
-    depends_build-append port:py25-simplejson
-    build.args      PYTHON=${prefix}/bin/python2.5
+    depends_build-append port:py26-simplejson
+    build.args      PYTHON=${prefix}/bin/python2.6
 }
 # no need for simplejson on Snow Leopard or higher
 
diff --git a/scripts/rabbitmq-server b/scripts/rabbitmq-server
index 0eb7092d..66ce4384 100755
--- a/scripts/rabbitmq-server
+++ b/scripts/rabbitmq-server
@@ -89,13 +89,10 @@ RABBITMQ_EBIN_ROOT="${RABBITMQ_HOME}/ebin"
 if [ "x" = "x$RABBITMQ_NODE_ONLY" ]; then
     if erl \
 	-pa "$RABBITMQ_EBIN_ROOT" \
-	-rabbit plugins_dir "\"$RABBITMQ_PLUGINS_DIR\"" \
-	-rabbit plugins_expand_dir "\"$RABBITMQ_PLUGINS_EXPAND_DIR\"" \
-	-rabbit rabbit_ebin "\"$RABBITMQ_EBIN_ROOT\"" \
 	-noinput \
 	-hidden \
-	-s rabbit_plugin_activator \
-	-extra "$@"
+	-s rabbit_prelaunch \
+	-extra "$RABBITMQ_PLUGINS_DIR" "${RABBITMQ_PLUGINS_EXPAND_DIR}" "${RABBITMQ_NODENAME}"
     then
 	RABBITMQ_BOOT_FILE="${RABBITMQ_PLUGINS_EXPAND_DIR}/rabbit"
 	RABBITMQ_EBIN_PATH=""
diff --git a/scripts/rabbitmq-server.bat b/scripts/rabbitmq-server.bat
index bd4120fa..872c87e3 100644
--- a/scripts/rabbitmq-server.bat
+++ b/scripts/rabbitmq-server.bat
@@ -117,15 +117,13 @@ set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin
 "!ERLANG_HOME!\bin\erl.exe" ^

 -pa "!RABBITMQ_EBIN_ROOT!" ^

 -noinput -hidden ^

--s rabbit_plugin_activator ^

--rabbit plugins_dir \""!RABBITMQ_PLUGINS_DIR:\=/!"\" ^

--rabbit plugins_expand_dir \""!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!"\" ^

--rabbit rabbit_ebin  \""!RABBITMQ_EBIN_ROOT:\=/!"\" ^

--extra !STAR!

-

-set RABBITMQ_BOOT_FILE=!RABBITMQ_MNESIA_DIR!\plugins-scratch\rabbit

-if not exist "!RABBITMQ_BOOT_FILE!.boot" (

-    echo Custom Boot File "!RABBITMQ_BOOT_FILE!.boot" is missing.

+-s rabbit_prelaunch ^

+-extra "!RABBITMQ_PLUGINS_DIR:\=/!" ^

+       "!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!" ^

+       "!RABBITMQ_NODENAME!"

+

+set RABBITMQ_BOOT_FILE=!RABBITMQ_PLUGINS_EXPAND_DIR!\rabbit

+if ERRORLEVEL 1 (

     exit /B 1

 )

 

diff --git a/scripts/rabbitmq-service.bat b/scripts/rabbitmq-service.bat
index ff25b146..d2592931 100644
--- a/scripts/rabbitmq-service.bat
+++ b/scripts/rabbitmq-service.bat
@@ -186,15 +186,13 @@ set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin
 "!ERLANG_HOME!\bin\erl.exe" ^

 -pa "!RABBITMQ_EBIN_ROOT!" ^

 -noinput -hidden ^

--s rabbit_plugin_activator ^

--rabbit plugins_dir \""!RABBITMQ_PLUGINS_DIR:\=/!"\" ^

--rabbit plugins_expand_dir \""!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!"\" ^

--rabbit rabbit_ebin  \""!RABBITMQ_EBIN_ROOT:\=/!"\" ^

--extra !STAR!

-

-set RABBITMQ_BOOT_FILE=!RABBITMQ_MNESIA_DIR!\plugins-scratch\rabbit

-if not exist "!RABBITMQ_BOOT_FILE!.boot" (

-    echo Custom Boot File "!RABBITMQ_BOOT_FILE!.boot" is missing.

+-s rabbit_prelaunch ^

+-extra "!RABBITMQ_PLUGINS_DIR:\=/!" ^

+       "!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!" ^

+       ""

+

+set RABBITMQ_BOOT_FILE=!RABBITMQ_PLUGINS_EXPAND_DIR!\rabbit

+if ERRORLEVEL 1 (

     exit /B 1

 )

 

diff --git a/src/rabbit.erl b/src/rabbit.erl
index a1dd2c2e..ace8f286 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -170,12 +170,6 @@
 
 %%---------------------------------------------------------------------------
 
--import(application).
--import(mnesia).
--import(lists).
--import(inet).
--import(gen_tcp).
-
 -include("rabbit_framing.hrl").
 -include("rabbit.hrl").
 
@@ -291,11 +285,11 @@ run_boot_step({StepName, Attributes}) ->
             io:format("-- ~s~n", [Description]);
         MFAs ->
             io:format("starting ~-60s ...", [Description]),
-            [case catch apply(M,F,A) of
-                 {'EXIT', Reason} ->
-                     boot_error("FAILED~nReason: ~p~n", [Reason]);
-                 ok ->
-                     ok
+            [try
+                 apply(M,F,A)
+             catch
+                 _:Reason -> boot_error("FAILED~nReason: ~p~nStacktrace: ~p~n",
+                                        [Reason, erlang:get_stacktrace()])
              end || {M,F,A} <- MFAs],
             io:format("done~n"),
             ok
@@ -315,43 +309,43 @@ edges(_Module, Steps) ->
             {Key, OtherStep} <- Atts,
             Key =:= requires orelse Key =:= enables].
 
-graph_build_error({vertex, duplicate, StepName}) ->
-    boot_error("Duplicate boot step name: ~w~n", [StepName]);
-graph_build_error({edge, Reason, From, To}) ->
-    boot_error(
-      "Could not add boot step dependency of ~w on ~w:~n~s",
-      [To, From,
-       case Reason of
-           {bad_vertex, V} ->
-               io_lib:format("Boot step not registered: ~w~n", [V]);
-           {bad_edge, [First | Rest]} ->
-               [io_lib:format("Cyclic dependency: ~w", [First]),
-                [io_lib:format(" depends on ~w", [Next]) || Next <- Rest],
-                io_lib:format(" depends on ~w~n", [First])]
-       end]).
-
 sort_boot_steps(UnsortedSteps) ->
-    G = rabbit_misc:build_acyclic_graph(
-          fun vertices/2, fun edges/2, fun graph_build_error/1, UnsortedSteps),
-
-    %% Use topological sort to find a consistent ordering (if there is
-    %% one, otherwise fail).
-    SortedStepsRev = [begin
-                          {StepName, Step} = digraph:vertex(G, StepName),
-                          Step
-                      end || StepName <- digraph_utils:topsort(G)],
-    SortedSteps = lists:reverse(SortedStepsRev),
-
-    digraph:delete(G),
-
-    %% Check that all mentioned {M,F,A} triples are exported.
-    case [{StepName, {M,F,A}}
-          || {StepName, Attributes} <- SortedSteps,
-             {mfa, {M,F,A}} <- Attributes,
-             not erlang:function_exported(M, F, length(A))] of
-        []               -> SortedSteps;
-        MissingFunctions -> boot_error("Boot step functions not exported: ~p~n",
-                                       [MissingFunctions])
+    case rabbit_misc:build_acyclic_graph(fun vertices/2, fun edges/2,
+                                         UnsortedSteps) of
+        {ok, G} ->
+            %% Use topological sort to find a consistent ordering (if
+            %% there is one, otherwise fail).
+            SortedSteps = lists:reverse(
+                            [begin
+                                 {StepName, Step} = digraph:vertex(G, StepName),
+                                 Step
+                             end || StepName <- digraph_utils:topsort(G)]),
+            digraph:delete(G),
+            %% Check that all mentioned {M,F,A} triples are exported.
+            case [{StepName, {M,F,A}} ||
+                     {StepName, Attributes} <- SortedSteps,
+                     {mfa, {M,F,A}}         <- Attributes,
+                     not erlang:function_exported(M, F, length(A))] of
+                []               -> SortedSteps;
+                MissingFunctions -> boot_error(
+                                      "Boot step functions not exported: ~p~n",
+                                      [MissingFunctions])
+            end;
+        {error, {vertex, duplicate, StepName}} ->
+            boot_error("Duplicate boot step name: ~w~n", [StepName]);
+        {error, {edge, Reason, From, To}} ->
+            boot_error(
+              "Could not add boot step dependency of ~w on ~w:~n~s",
+              [To, From,
+               case Reason of
+                   {bad_vertex, V} ->
+                       io_lib:format("Boot step not registered: ~w~n", [V]);
+                   {bad_edge, [First | Rest]} ->
+                       [io_lib:format("Cyclic dependency: ~w", [First]),
+                        [io_lib:format(" depends on ~w", [Next]) ||
+                            Next <- Rest],
+                        io_lib:format(" depends on ~w~n", [First])]
+               end])
     end.
 
 %%---------------------------------------------------------------------------
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 5cdd0e3c..775c631d 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -34,6 +34,7 @@
 -export([start/0, stop/0, declare/5, delete_immediately/1, delete/3, purge/1]).
 -export([internal_declare/2, internal_delete/1,
          maybe_run_queue_via_backing_queue/2,
+         maybe_run_queue_via_backing_queue_async/2,
          update_ram_duration/1, set_ram_duration_target/2,
          set_maximum_since_use/2, maybe_expire/1, drop_expired/1]).
 -export([pseudo_queue/2]).
@@ -48,11 +49,6 @@
 -export([commit_all/3, rollback_all/3, notify_down_all/2, limit_all/3]).
 -export([on_node_down/1]).
 
--import(mnesia).
--import(gen_server2).
--import(lists).
--import(queue).
-
 -include("rabbit.hrl").
 -include_lib("stdlib/include/qlc.hrl").
 
@@ -156,7 +152,9 @@
         (name()) -> rabbit_types:ok_or_error('not_found') |
                     rabbit_types:connection_exit()).
 -spec(maybe_run_queue_via_backing_queue/2 ::
-        (pid(), (fun ((A) -> A))) -> 'ok').
+        (pid(), (fun ((A) -> A | {any(), A}))) -> 'ok').
+-spec(maybe_run_queue_via_backing_queue_async/2 ::
+        (pid(), (fun ((A) -> A | {any(), A}))) -> 'ok').
 -spec(update_ram_duration/1 :: (pid()) -> 'ok').
 -spec(set_ram_duration_target/2 :: (pid(), number() | 'infinity') -> 'ok').
 -spec(set_maximum_since_use/2 :: (pid(), non_neg_integer()) -> 'ok').
@@ -279,7 +277,7 @@ assert_equivalence(#amqqueue{durable     = Durable,
 assert_equivalence(#amqqueue{name = QueueName},
                    _Durable, _AutoDelete, _RequiredArgs, _Owner) ->
     rabbit_misc:protocol_error(
-      not_allowed, "parameters for ~s not equivalent",
+      precondition_failed, "parameters for ~s not equivalent",
       [rabbit_misc:rs(QueueName)]).
 
 check_exclusive_access(Q, Owner) -> check_exclusive_access(Q, Owner, lax).
@@ -380,16 +378,13 @@ delete(#amqqueue{ pid = QPid }, IfUnused, IfEmpty) ->
 
 purge(#amqqueue{ pid = QPid }) -> delegate_call(QPid, purge, infinity).
 
-deliver(QPid, #delivery{immediate = true,
-                        txn = Txn, sender = ChPid, message = Message}) ->
-    gen_server2:call(QPid, {deliver_immediately, Txn, Message, ChPid},
-                     infinity);
-deliver(QPid, #delivery{mandatory = true,
-                        txn = Txn, sender = ChPid, message = Message}) ->
-    gen_server2:call(QPid, {deliver, Txn, Message, ChPid}, infinity),
+deliver(QPid, Delivery = #delivery{immediate = true}) ->
+    gen_server2:call(QPid, {deliver_immediately, Delivery}, infinity);
+deliver(QPid, Delivery = #delivery{mandatory = true}) ->
+    gen_server2:call(QPid, {deliver, Delivery}, infinity),
     true;
-deliver(QPid, #delivery{txn = Txn, sender = ChPid, message = Message}) ->
-    gen_server2:cast(QPid, {deliver, Txn, Message, ChPid}),
+deliver(QPid, Delivery) ->
+    gen_server2:cast(QPid, {deliver, Delivery}),
     true.
 
 requeue(QPid, MsgIds, ChPid) ->
@@ -466,6 +461,9 @@ internal_delete(QueueName) ->
 maybe_run_queue_via_backing_queue(QPid, Fun) ->
     gen_server2:call(QPid, {maybe_run_queue_via_backing_queue, Fun}, infinity).
 
+maybe_run_queue_via_backing_queue_async(QPid, Fun) ->
+    gen_server2:cast(QPid, {maybe_run_queue_via_backing_queue, Fun}).
+
 update_ram_duration(QPid) ->
     gen_server2:cast(QPid, update_ram_duration).
 
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 75f285df..78bb6835 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -39,7 +39,8 @@
 -define(SYNC_INTERVAL,                 5). %% milliseconds
 -define(RAM_DURATION_UPDATE_INTERVAL,  5000).
 
--define(BASE_MESSAGE_PROPERTIES, #message_properties{expiry = undefined}).
+-define(BASE_MESSAGE_PROPERTIES,
+        #message_properties{expiry = undefined, needs_confirming = false}).
 
 -export([start_link/1, info_keys/0]).
 
@@ -47,10 +48,6 @@
          handle_info/2, handle_pre_hibernate/1, prioritise_call/3,
          prioritise_cast/2, prioritise_info/2]).
 
--import(queue).
--import(erlang).
--import(lists).
-
 % Queue's state
 -record(q, {q,
             exclusive_consumer,
@@ -64,6 +61,7 @@
             rate_timer_ref,
             expiry_timer_ref,
             stats_timer,
+            guid_to_channel,
             ttl,
             ttl_timer_ref
            }).
@@ -128,7 +126,8 @@ init(Q) ->
             rate_timer_ref      = undefined,
             expiry_timer_ref    = undefined,
             ttl                 = undefined,
-            stats_timer         = rabbit_event:init_stats_timer()}, hibernate,
+            stats_timer         = rabbit_event:init_stats_timer(),
+            guid_to_channel     = dict:new()}, hibernate,
      {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
 
 terminate(shutdown,      State = #q{backing_queue = BQ}) ->
@@ -373,11 +372,13 @@ deliver_msgs_to_consumers(Funs = {PredFun, DeliverFun}, FunAcc,
                     rabbit_channel:deliver(
                       ChPid, ConsumerTag, AckRequired,
                       {QName, self(), AckTag, IsDelivered, Message}),
-                    ChAckTags1 = case AckRequired of
-                                     true  -> sets:add_element(
-                                                AckTag, ChAckTags);
-                                     false -> ChAckTags
-                                 end,
+                    {State2, ChAckTags1} =
+                        case AckRequired of
+                            true  -> {State1,
+                                      sets:add_element(AckTag, ChAckTags)};
+                            false -> {confirm_message(Message, State1),
+                                      ChAckTags}
+                        end,
                     NewC = C#cr{unsent_message_count = Count + 1,
                                 acktags = ChAckTags1},
                     true = maybe_store_ch_record(NewC),
@@ -393,10 +394,10 @@ deliver_msgs_to_consumers(Funs = {PredFun, DeliverFun}, FunAcc,
                                 {ActiveConsumers1,
                                  queue:in(QEntry, BlockedConsumers1)}
                         end,
-                    State2 = State1#q{
+                    State3 = State2#q{
                                active_consumers = NewActiveConsumers,
                                blocked_consumers = NewBlockedConsumers},
-                    deliver_msgs_to_consumers(Funs, FunAcc1, State2);
+                    deliver_msgs_to_consumers(Funs, FunAcc1, State3);
                 %% if IsMsgReady then we've hit the limiter
                 false when IsMsgReady ->
                     true = maybe_store_ch_record(C#cr{is_limit_active = true}),
@@ -424,6 +425,39 @@ deliver_from_queue_deliver(AckRequired, false, State) ->
         fetch(AckRequired, State),
     {{Message, IsDelivered, AckTag}, 0 == Remaining, State1}.
 
+confirm_messages(Guids, State) ->
+    lists:foldl(fun confirm_message_by_guid/2, State, Guids).
+
+confirm_message_by_guid(Guid, State = #q{guid_to_channel = GTC}) ->
+    case dict:find(Guid, GTC) of
+        {ok, {_    , undefined}} -> ok;
+        {ok, {ChPid, MsgSeqNo}}  -> rabbit_channel:confirm(ChPid, MsgSeqNo);
+        _                        -> ok
+    end,
+    State#q{guid_to_channel = dict:erase(Guid, GTC)}.
+
+confirm_message(#basic_message{guid = Guid}, State) ->
+    confirm_message_by_guid(Guid, State).
+
+record_confirm_message(#delivery{msg_seq_no = undefined}, State) ->
+    State;
+record_confirm_message(#delivery{sender     = ChPid,
+                                 msg_seq_no = MsgSeqNo,
+                                 message    = #basic_message {
+                                   is_persistent = true,
+                                   guid          = Guid}},
+                       State =
+                           #q{guid_to_channel = GTC,
+                              q               = #amqqueue{durable = true}}) ->
+    State#q{guid_to_channel = dict:store(Guid, {ChPid, MsgSeqNo}, GTC)};
+record_confirm_message(_Delivery, State) ->
+    State.
+
+ack_by_acktags(AckTags, State = #q{backing_queue       = BQ,
+                                   backing_queue_state = BQS}) ->
+    {AckdGuids, BQS1} = BQ:ack(AckTags, BQS),
+    confirm_messages(AckdGuids, State#q{backing_queue_state = BQS1}).
+
 run_message_queue(State) ->
     Funs = {fun deliver_from_queue_pred/2,
             fun deliver_from_queue_deliver/3},
@@ -433,7 +467,17 @@ run_message_queue(State) ->
     {_IsEmpty1, State2} = deliver_msgs_to_consumers(Funs, IsEmpty, State1),
     State2.
 
-attempt_delivery(none, _ChPid, Message, State = #q{backing_queue = BQ}) ->
+attempt_delivery(#delivery{txn        = none,
+                           sender     = ChPid,
+                           message    = Message,
+                           msg_seq_no = MsgSeqNo},
+                 State = #q{backing_queue = BQ, q = Q}) ->
+    NeedsConfirming = Message#basic_message.is_persistent andalso
+                      Q#amqqueue.durable,
+    case NeedsConfirming of
+        false -> rabbit_channel:confirm(ChPid, MsgSeqNo);
+        _     -> ok
+    end,
     PredFun = fun (IsEmpty, _State) -> not IsEmpty end,
     DeliverFun =
         fun (AckRequired, false, State1 = #q{backing_queue_state = BQS}) ->
@@ -441,29 +485,36 @@ attempt_delivery(none, _ChPid, Message, State = #q{backing_queue = BQ}) ->
                 %% not being enqueued, so we use an empty
                 %% message_properties.
                 {AckTag, BQS1} =
-                    BQ:publish_delivered(AckRequired, Message,
-                                         ?BASE_MESSAGE_PROPERTIES, BQS),
+                    BQ:publish_delivered(
+                      AckRequired, Message,
+                      (?BASE_MESSAGE_PROPERTIES)#message_properties{
+                        needs_confirming = NeedsConfirming},
+                      BQS),
                 {{Message, false, AckTag}, true,
                  State1#q{backing_queue_state = BQS1}}
         end,
     deliver_msgs_to_consumers({ PredFun, DeliverFun }, false, State);
-attempt_delivery(Txn, ChPid, Message, State = #q{backing_queue       = BQ,
-                                                 backing_queue_state = BQS}) ->
+attempt_delivery(#delivery{txn = Txn,
+                           sender  = ChPid,
+                           message = Message},
+                 State = #q{backing_queue = BQ,
+                            backing_queue_state = BQS}) ->
     record_current_channel_tx(ChPid, Txn),
     {true,
      State#q{backing_queue_state =
                  BQ:tx_publish(Txn, Message, ?BASE_MESSAGE_PROPERTIES, BQS)}}.
 
-deliver_or_enqueue(Txn, ChPid, Message, State = #q{backing_queue = BQ}) ->
-    case attempt_delivery(Txn, ChPid, Message, State) of
-        {true, NewState} ->
-            {true, NewState};
-        {false, NewState} ->
-            %% Txn is none and no unblocked channels with consumers
-            BQS = BQ:publish(Message,
-                             message_properties(State),
-                             State #q.backing_queue_state),
-            {false, ensure_ttl_timer(NewState#q{backing_queue_state = BQS})}
+deliver_or_enqueue(Delivery, State) ->
+    case attempt_delivery(Delivery, record_confirm_message(Delivery, State)) of
+        {true, State1} ->
+            {true, State1};
+        {false, State1 = #q{backing_queue = BQ, backing_queue_state = BQS}} ->
+            #delivery{message = Message, msg_seq_no = MsgSeqNo} = Delivery,
+            BQS1 = BQ:publish(Message,
+                              (message_properties(State)) #message_properties{
+                                needs_confirming = (MsgSeqNo =/= undefined)},
+                              BQS),
+            {false, ensure_ttl_timer(State1#q{backing_queue_state = BQS1})}
     end.
 
 requeue_and_run(AckTags, State = #q{backing_queue = BQ, ttl=TTL}) ->
@@ -566,7 +617,12 @@ maybe_send_reply(ChPid, Msg) -> ok = rabbit_channel:send_command(ChPid, Msg).
 qname(#q{q = #amqqueue{name = QName}}) -> QName.
 
 maybe_run_queue_via_backing_queue(Fun, State = #q{backing_queue_state = BQS}) ->
-    run_message_queue(State#q{backing_queue_state = Fun(BQS)}).
+    {BQS2, State1} =
+        case Fun(BQS) of
+            {{confirm, Guids}, BQS1} -> {BQS1, confirm_messages(Guids, State)};
+            BQS1                     -> {BQS1, State}
+        end,
+    run_message_queue(State1#q{backing_queue_state = BQS2}).
 
 commit_transaction(Txn, From, ChPid, State = #q{backing_queue       = BQ,
                                                 backing_queue_state = BQS,
@@ -669,7 +725,10 @@ i(Item, _) ->
     throw({bad_argument, Item}).
 
 emit_stats(State) ->
-    rabbit_event:notify(queue_stats, infos(?STATISTICS_KEYS, State)).
+    emit_stats(State, []).
+
+emit_stats(State, Extra) ->
+    rabbit_event:notify(queue_stats, Extra ++ infos(?STATISTICS_KEYS, State)).
 
 %---------------------------------------------------------------------------
 
@@ -742,7 +801,8 @@ handle_call(consumers, _From,
                     [{ChPid, ConsumerTag, AckRequired} | Acc]
             end, [], queue:join(ActiveConsumers, BlockedConsumers)), State);
 
-handle_call({deliver_immediately, Txn, Message, ChPid}, _From, State) ->
+handle_call({deliver_immediately, Delivery = #delivery{message = Message}},
+            _From, State) ->
     %% Synchronous, "immediate" delivery mode
     %%
     %% FIXME: Is this correct semantics?
@@ -756,12 +816,16 @@ handle_call({deliver_immediately, Txn, Message, ChPid}, _From, State) ->
     %% just all ready-to-consume queues get the message, with unready
     %% queues discarding the message?
     %%
-    {Delivered, NewState} = attempt_delivery(Txn, ChPid, Message, State),
-    reply(Delivered, NewState);
-
-handle_call({deliver, Txn, Message, ChPid}, _From, State) ->
+    {Delivered, State1} =
+        attempt_delivery(Delivery, record_confirm_message(Delivery, State)),
+    reply(Delivered, case Delivered of
+                         true  -> State1;
+                         false -> confirm_message(Message, State1)
+                     end);
+
+handle_call({deliver, Delivery}, _From, State) ->
     %% Synchronous, "mandatory" delivery mode
-    {Delivered, NewState} = deliver_or_enqueue(Txn, ChPid, Message, State),
+    {Delivered, NewState} = deliver_or_enqueue(Delivery, State),
     reply(Delivered, NewState);
 
 handle_call({commit, Txn, ChPid}, From, State) ->
@@ -787,15 +851,18 @@ handle_call({basic_get, ChPid, NoAck}, _From,
         {empty, State2} ->
             reply(empty, State2);
         {{Message, IsDelivered, AckTag, Remaining}, State2} ->
-            case AckRequired of
-                true  -> C = #cr{acktags = ChAckTags} = ch_record(ChPid),
-                         true = maybe_store_ch_record(
-                                  C#cr{acktags = sets:add_element(AckTag,
-                                                                  ChAckTags)});
-                false -> ok
-            end,
+            State3 =
+                case AckRequired of
+                    true  -> C = #cr{acktags = ChAckTags} = ch_record(ChPid),
+                             true = maybe_store_ch_record(
+                                      C#cr{acktags =
+                                               sets:add_element(AckTag,
+                                                                ChAckTags)}),
+                             State2;
+                    false -> confirm_message(Message, State2)
+                end,
             Msg = {QName, self(), AckTag, IsDelivered, Message},
-            reply({ok, Remaining, Msg}, State2)
+            reply({ok, Remaining, Msg}, State3)
     end;
 
 handle_call({basic_consume, NoAck, ChPid, LimiterPid,
@@ -908,9 +975,13 @@ handle_call({requeue, AckTags, ChPid}, From, State) ->
 handle_call({maybe_run_queue_via_backing_queue, Fun}, _From, State) ->
     reply(ok, maybe_run_queue_via_backing_queue(Fun, State)).
 
-handle_cast({deliver, Txn, Message, ChPid}, State) ->
+
+handle_cast({maybe_run_queue_via_backing_queue, Fun}, State) ->
+    noreply(maybe_run_queue_via_backing_queue(Fun, State));
+
+handle_cast({deliver, Delivery}, State) ->
     %% Asynchronous, non-"mandatory", non-"immediate" deliver mode.
-    {_Delivered, NewState} = deliver_or_enqueue(Txn, ChPid, Message, State),
+    {_Delivered, NewState} = deliver_or_enqueue(Delivery, State),
     noreply(NewState);
 
 handle_cast({ack, Txn, AckTags, ChPid},
@@ -919,18 +990,21 @@ handle_cast({ack, Txn, AckTags, ChPid},
         not_found ->
             noreply(State);
         C = #cr{acktags = ChAckTags} ->
-            {C1, BQS1} =
+            {C1, State1} =
                 case Txn of
                     none -> ChAckTags1 = subtract_acks(ChAckTags, AckTags),
-                            {C#cr{acktags = ChAckTags1}, BQ:ack(AckTags, BQS)};
-                    _    -> {C#cr{txn = Txn}, BQ:tx_ack(Txn, AckTags, BQS)}
+                            NewC = C#cr{acktags = ChAckTags1},
+                            NewState = ack_by_acktags(AckTags, State),
+                            {NewC, NewState};
+                    _    -> BQS1 = BQ:tx_ack(Txn, AckTags, BQS),
+                            {C#cr{txn = Txn},
+                             State#q{backing_queue_state = BQS1}}
                 end,
             maybe_store_ch_record(C1),
-            noreply(State#q{backing_queue_state = BQS1})
+            noreply(State1)
     end;
 
-handle_cast({reject, AckTags, Requeue, ChPid},
-            State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
+handle_cast({reject, AckTags, Requeue, ChPid}, State) ->
     case lookup_ch(ChPid) of
         not_found ->
             noreply(State);
@@ -939,8 +1013,7 @@ handle_cast({reject, AckTags, Requeue, ChPid},
             maybe_store_ch_record(C#cr{acktags = ChAckTags1}),
             noreply(case Requeue of
                         true  -> requeue_and_run(AckTags, State);
-                        false -> BQS1 = BQ:ack(AckTags, BQS),
-                                 State #q { backing_queue_state = BQS1 }
+                        false -> ack_by_acktags(AckTags, State)
                     end)
     end;
 
@@ -1053,7 +1126,10 @@ handle_pre_hibernate(State = #q{backing_queue = BQ,
     DesiredDuration =
         rabbit_memory_monitor:report_ram_duration(self(), infinity),
     BQS2 = BQ:set_ram_duration_target(DesiredDuration, BQS1),
-    rabbit_event:if_enabled(StatsTimer, fun () -> emit_stats(State) end),
+    rabbit_event:if_enabled(StatsTimer,
+                            fun () ->
+                                    emit_stats(State, [{idle_since, now()}])
+                            end),
     State1 = State#q{stats_timer = rabbit_event:stop_stats_timer(StatsTimer),
                      backing_queue_state = BQS2},
     {hibernate, stop_rate_timer(State1)}.
diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
index 38412982..1ac39b65 100644
--- a/src/rabbit_basic.erl
+++ b/src/rabbit_basic.erl
@@ -33,7 +33,7 @@
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
 
--export([publish/1, message/4, properties/1, delivery/4]).
+-export([publish/1, message/4, properties/1, delivery/5]).
 -export([publish/4, publish/7]).
 -export([build_content/2, from_content/1]).
 -export([is_message_persistent/1]).
@@ -50,9 +50,10 @@
 
 -spec(publish/1 ::
         (rabbit_types:delivery()) -> publish_result()).
--spec(delivery/4 ::
+-spec(delivery/5 ::
         (boolean(), boolean(), rabbit_types:maybe(rabbit_types:txn()),
-         rabbit_types:message()) -> rabbit_types:delivery()).
+         rabbit_types:message(), undefined | integer()) ->
+                         rabbit_types:delivery()).
 -spec(message/4 ::
         (rabbit_exchange:name(), rabbit_router:routing_key(),
          properties_input(), binary()) ->
@@ -88,9 +89,9 @@ publish(Delivery = #delivery{
             Other
     end.
 
-delivery(Mandatory, Immediate, Txn, Message) ->
+delivery(Mandatory, Immediate, Txn, Message, MsgSeqNo) ->
     #delivery{mandatory = Mandatory, immediate = Immediate, txn = Txn,
-              sender = self(), message = Message}.
+              sender = self(), message = Message, msg_seq_no = MsgSeqNo}.
 
 build_content(Properties, BodyBin) ->
     %% basic.publish hasn't changed so we can just hard-code amqp_0_9_1
@@ -157,7 +158,8 @@ publish(ExchangeName, RoutingKeyBin, Mandatory, Immediate, Txn, Properties,
         BodyBin) ->
     publish(delivery(Mandatory, Immediate, Txn,
                      message(ExchangeName, RoutingKeyBin,
-                             properties(Properties), BodyBin))).
+                             properties(Properties), BodyBin),
+                     undefined)).
 
 is_message_persistent(#content{properties = #'P_basic'{
                                  delivery_mode = Mode}}) ->
diff --git a/src/rabbit_binary_generator.erl b/src/rabbit_binary_generator.erl
index b2997ae2..a5297a70 100644
--- a/src/rabbit_binary_generator.erl
+++ b/src/rabbit_binary_generator.erl
@@ -49,8 +49,6 @@
 -export([ensure_content_encoded/2, clear_encoded_content/1]).
 -export([map_exception/3]).
 
--import(lists).
-
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
diff --git a/src/rabbit_binary_parser.erl b/src/rabbit_binary_parser.erl
index ebf063f0..4b4358b4 100644
--- a/src/rabbit_binary_parser.erl
+++ b/src/rabbit_binary_parser.erl
@@ -36,8 +36,6 @@
 -export([parse_table/1, parse_properties/2]).
 -export([ensure_content_decoded/1, clear_decoded_content/1]).
 
--import(lists).
-
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
diff --git a/src/rabbit_binding.erl b/src/rabbit_binding.erl
index 9d1399f7..668fb9bb 100644
--- a/src/rabbit_binding.erl
+++ b/src/rabbit_binding.erl
@@ -350,10 +350,10 @@ group_bindings_fold(Fun, SrcName, Acc, Removed, Bindings) ->
     group_bindings_fold(Fun, Fun(SrcName, Bindings, Acc), Removed).
 
 maybe_auto_delete(XName, Bindings, Deletions) ->
-    case rabbit_exchange:lookup(XName) of
-        {error, not_found} ->
+    case mnesia:read(rabbit_exchange, XName) of
+        [] ->
             add_deletion(XName, {undefined, not_deleted, Bindings}, Deletions);
-        {ok, X} ->
+        [X] ->
             add_deletion(XName, {X, not_deleted, Bindings},
                          case rabbit_exchange:maybe_auto_delete(X) of
                              not_deleted           -> Deletions;
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index b46bf1b4..25f88a1b 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -38,7 +38,7 @@
 -export([start_link/7, do/2, do/3, shutdown/1]).
 -export([send_command/2, deliver/4, flushed/2]).
 -export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1]).
--export([emit_stats/1, flush/1]).
+-export([emit_stats/1, flush/1, flush_multiple_acks/1, confirm/2]).
 
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
          handle_info/2, handle_pre_hibernate/1, prioritise_call/3,
@@ -48,7 +48,9 @@
              start_limiter_fun, transaction_id, tx_participants, next_tag,
              uncommitted_ack_q, unacked_message_q,
              username, virtual_host, most_recently_declared_queue,
-             consumer_mapping, blocking, queue_collector_pid, stats_timer}).
+             consumer_mapping, blocking, queue_collector_pid, stats_timer,
+             confirm_enabled, publish_seqno, confirm_multiple, confirm_tref,
+             held_confirms, unconfirmed, queues_for_msg}).
 
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 
@@ -70,6 +72,8 @@
 
 -define(INFO_KEYS, ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [pid]).
 
+-define(FLUSH_MULTIPLE_ACKS_INTERVAL, 1000).
+
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
@@ -99,6 +103,8 @@
 -spec(info_all/0 :: () -> [rabbit_types:infos()]).
 -spec(info_all/1 :: (rabbit_types:info_keys()) -> [rabbit_types:infos()]).
 -spec(emit_stats/1 :: (pid()) -> 'ok').
+-spec(flush_multiple_acks/1 :: (pid()) -> 'ok').
+-spec(confirm/2 ::(pid(), non_neg_integer()) -> 'ok').
 
 -endif.
 
@@ -153,6 +159,12 @@ emit_stats(Pid) ->
 flush(Pid) ->
     gen_server2:call(Pid, flush).
 
+flush_multiple_acks(Pid) ->
+    gen_server2:cast(Pid, flush_multiple_acks).
+
+confirm(Pid, MsgSeqNo) ->
+    gen_server2:cast(Pid, {confirm, MsgSeqNo, self()}).
+
 %%---------------------------------------------------------------------------
 
 init([Channel, ReaderPid, WriterPid, Username, VHost, CollectorPid,
@@ -177,7 +189,13 @@ init([Channel, ReaderPid, WriterPid, Username, VHost, CollectorPid,
                 consumer_mapping        = dict:new(),
                 blocking                = dict:new(),
                 queue_collector_pid     = CollectorPid,
-                stats_timer             = StatsTimer},
+                stats_timer             = StatsTimer,
+                confirm_enabled         = false,
+                publish_seqno           = 0,
+                confirm_multiple        = false,
+                held_confirms           = gb_sets:new(),
+                unconfirmed             = gb_sets:new(),
+                queues_for_msg          = dict:new()},
     rabbit_event:notify(channel_created, infos(?CREATION_EVENT_KEYS, State)),
     rabbit_event:if_enabled(StatsTimer,
                             fun() -> internal_emit_stats(State) end),
@@ -242,12 +260,24 @@ handle_cast({command, Msg}, State = #ch{writer_pid = WriterPid}) ->
     ok = rabbit_writer:send_command(WriterPid, Msg),
     noreply(State);
 
-handle_cast({deliver, ConsumerTag, AckRequired, Msg},
+handle_cast({deliver, ConsumerTag, AckRequired,
+	     Msg = {_QName, QPid, _MsgId, Redelivered,
+		    #basic_message{exchange_name = ExchangeName,
+				   routing_key = RoutingKey,
+				   content = Content}}},
             State = #ch{writer_pid = WriterPid,
                         next_tag = DeliveryTag}) ->
-    State1 = lock_message(AckRequired, {DeliveryTag, ConsumerTag, Msg}, State),
-    ok = internal_deliver(WriterPid, true, ConsumerTag, DeliveryTag, Msg),
-    {_QName, QPid, _MsgId, _Redelivered, _Msg} = Msg,
+    State1 = lock_message(AckRequired,
+                          ack_record(DeliveryTag, ConsumerTag, Msg),
+                          State),
+
+    M = #'basic.deliver'{consumer_tag = ConsumerTag,
+                         delivery_tag = DeliveryTag,
+                         redelivered = Redelivered,
+                         exchange = ExchangeName#resource.name,
+                         routing_key = RoutingKey},
+    rabbit_writer:send_command_and_notify(WriterPid, QPid, self(), M, Content),
+
     maybe_incr_stats([{QPid, 1}],
                      case AckRequired of
                          true  -> deliver;
@@ -259,19 +289,38 @@ handle_cast(emit_stats, State = #ch{stats_timer = StatsTimer}) ->
     internal_emit_stats(State),
     {noreply,
      State#ch{stats_timer = rabbit_event:reset_stats_timer(StatsTimer)},
-     hibernate}.
-
-handle_info({'DOWN', _MRef, process, QPid, _Reason}, State) ->
+     hibernate};
+
+handle_cast(flush_multiple_acks, State) ->
+    {noreply, flush_multiple(State)};
+
+handle_cast({confirm, MsgSeqNo, From}, State) ->
+    {noreply, send_or_enqueue_ack(MsgSeqNo, From, State)}.
+
+handle_info({'DOWN', _MRef, process, QPid, _Reason},
+            State = #ch{queues_for_msg = QFM}) ->
+    State1 = dict:fold(
+               fun(Msg, QPids, State0 = #ch{queues_for_msg = QFM0}) ->
+                       Qs = sets:del_element(QPid, QPids),
+                       case sets:size(Qs) of
+                           0 -> send_or_enqueue_ack(Msg, QPid, State0);
+                           _ -> State0#ch{queues_for_msg =
+                                              dict:store(Msg, Qs, QFM0)}
+                       end
+               end, State, QFM),
     erase_queue_stats(QPid),
-    {noreply, queue_blocked(QPid, State), hibernate}.
+    {noreply, queue_blocked(QPid, State1), hibernate}.
 
-handle_pre_hibernate(State = #ch{stats_timer = StatsTimer}) ->
+handle_pre_hibernate(State = #ch{stats_timer   = StatsTimer}) ->
     ok = clear_permission_cache(),
-    rabbit_event:if_enabled(StatsTimer, fun () ->
-                                                internal_emit_stats(State)
-                                        end),
-    {hibernate,
-     State#ch{stats_timer = rabbit_event:stop_stats_timer(StatsTimer)}}.
+    State1 = flush_multiple(State),
+    rabbit_event:if_enabled(StatsTimer,
+                            fun () ->
+                                    internal_emit_stats(
+                                      State, [{idle_since, now()}])
+                            end),
+    StatsTimer1 = rabbit_event:stop_stats_timer(StatsTimer),
+    {hibernate, State1#ch{stats_timer = StatsTimer1}}.
 
 terminate(_Reason, State = #ch{state = terminating}) ->
     terminate(State);
@@ -427,6 +476,52 @@ queue_blocked(QPid, State = #ch{blocking = Blocking}) ->
                       State#ch{blocking = Blocking1}
     end.
 
+send_or_enqueue_ack(undefined, _QPid, State) ->
+    State;
+send_or_enqueue_ack(_MsgSeqNo, _QPid, State = #ch{confirm_enabled = false}) ->
+    State;
+send_or_enqueue_ack(MsgSeqNo, QPid, State = #ch{confirm_multiple = false}) ->
+    do_if_unconfirmed(MsgSeqNo, QPid,
+                      fun(MSN, State1 = #ch{writer_pid = WriterPid}) ->
+                              ok = rabbit_writer:send_command(
+                                     WriterPid, #'basic.ack'{
+                                       delivery_tag = MSN}),
+                              State1
+                      end, State);
+send_or_enqueue_ack(MsgSeqNo, QPid, State = #ch{confirm_multiple = true}) ->
+    do_if_unconfirmed(MsgSeqNo, QPid,
+                      fun(MSN, State1 = #ch{held_confirms = As}) ->
+                              start_confirm_timer(
+                                State1#ch{held_confirms = gb_sets:add(MSN, As)})
+                      end, State).
+
+do_if_unconfirmed(MsgSeqNo, QPid, ConfirmFun,
+                  State = #ch{unconfirmed    = UC,
+                              queues_for_msg = QFM}) ->
+    %% clears references to MsgSeqNo and does ConfirmFun
+    case gb_sets:is_element(MsgSeqNo, UC) of
+        true  ->
+            Unconfirmed1 = gb_sets:delete(MsgSeqNo, UC),
+            case QPid of
+                undefined ->
+                    ConfirmFun(MsgSeqNo, State#ch{unconfirmed = Unconfirmed1});
+                _  ->
+                    {ok, Qs} = dict:find(MsgSeqNo, QFM),
+                    Qs1 = sets:del_element(QPid, Qs),
+                    case sets:size(Qs1) of
+                        0 -> ConfirmFun(MsgSeqNo,
+                                        State#ch{
+                                          queues_for_msg =
+                                              dict:erase(MsgSeqNo, QFM),
+                                          unconfirmed = Unconfirmed1});
+                        _ -> State#ch{queues_for_msg =
+                                          dict:store(MsgSeqNo, Qs1, QFM)}
+                    end
+            end;
+        false ->
+            State
+    end.
+
 handle_method(#'channel.open'{}, _, State = #ch{state = starting}) ->
     {reply, #'channel.open_ok'{}, State#ch{state = running}};
 
@@ -449,9 +544,9 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                routing_key = RoutingKey,
                                mandatory   = Mandatory,
                                immediate   = Immediate},
-              Content, State = #ch{virtual_host   = VHostPath,
-                                   transaction_id = TxnKey,
-                                   writer_pid     = WriterPid}) ->
+              Content, State = #ch{virtual_host    = VHostPath,
+                                   transaction_id  = TxnKey,
+                                   confirm_enabled = ConfirmEnabled}) ->
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_write_permitted(ExchangeName, State),
     Exchange = rabbit_exchange:lookup_or_die(ExchangeName),
@@ -460,6 +555,15 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     %% certain to want to look at delivery-mode and priority.
     DecodedContent = rabbit_binary_parser:ensure_content_decoded(Content),
     IsPersistent = is_message_persistent(DecodedContent),
+    {MsgSeqNo, State1}
+        = case ConfirmEnabled of
+              false -> {undefined, State};
+              true  -> SeqNo = State#ch.publish_seqno,
+                       {SeqNo,
+                        State#ch{publish_seqno = SeqNo + 1,
+                                 unconfirmed =
+                                     gb_sets:add(SeqNo, State#ch.unconfirmed)}}
+          end,
     Message = #basic_message{exchange_name = ExchangeName,
                              routing_key   = RoutingKey,
                              content       = DecodedContent,
@@ -468,18 +572,16 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     {RoutingRes, DeliveredQPids} =
         rabbit_exchange:publish(
           Exchange,
-          rabbit_basic:delivery(Mandatory, Immediate, TxnKey, Message)),
-    case RoutingRes of
-        routed        -> ok;
-        unroutable    -> ok = basic_return(Message, WriterPid, no_route);
-        not_delivered -> ok = basic_return(Message, WriterPid, no_consumers)
-    end,
+          rabbit_basic:delivery(Mandatory, Immediate, TxnKey, Message,
+                                MsgSeqNo)),
+    State2 = process_routing_result(RoutingRes, DeliveredQPids,
+                                    MsgSeqNo, Message, State1),
     maybe_incr_stats([{ExchangeName, 1} |
                       [{{QPid, ExchangeName}, 1} ||
-                          QPid <- DeliveredQPids]], publish, State),
+                          QPid <- DeliveredQPids]], publish, State2),
     {noreply, case TxnKey of
-                  none -> State;
-                  _    -> add_tx_participants(DeliveredQPids, State)
+                  none -> State2;
+                  _    -> add_tx_participants(DeliveredQPids, State2)
               end};
 
 handle_method(#'basic.ack'{delivery_tag = DeliveryTag,
@@ -516,7 +618,9 @@ handle_method(#'basic.get'{queue = QueueNameBin,
                 #basic_message{exchange_name = ExchangeName,
                                routing_key = RoutingKey,
                                content = Content}}} ->
-            State1 = lock_message(not(NoAck), {DeliveryTag, none, Msg}, State),
+            State1 = lock_message(not(NoAck),
+                                  ack_record(DeliveryTag, none, Msg),
+                                  State),
             maybe_incr_stats([{QPid, 1}],
                              case NoAck of
                                  true  -> get_no_ack;
@@ -652,30 +756,8 @@ handle_method(#'basic.recover_async'{requeue = true},
     %% variant of this method
     {noreply, State#ch{unacked_message_q = queue:new()}};
 
-handle_method(#'basic.recover_async'{requeue = false},
-              _, State = #ch{writer_pid = WriterPid,
-                             unacked_message_q = UAMQ}) ->
-    ok = rabbit_misc:queue_fold(
-           fun ({_DeliveryTag, none, _Msg}, ok) ->
-                   %% Was sent as a basic.get_ok. Don't redeliver
-                   %% it. FIXME: appropriate?
-                   ok;
-               ({DeliveryTag, ConsumerTag,
-                 {QName, QPid, MsgId, _Redelivered, Message}}, ok) ->
-                   %% Was sent as a proper consumer delivery.  Resend
-                   %% it as before.
-                   %%
-                   %% FIXME: What should happen if the consumer's been
-                   %% cancelled since?
-                   %%
-                   %% FIXME: should we allocate a fresh DeliveryTag?
-                   internal_deliver(
-                     WriterPid, false, ConsumerTag, DeliveryTag,
-                     {QName, QPid, MsgId, true, Message})
-           end, ok, UAMQ),
-    %% No answer required - basic.recover is the newer, synchronous
-    %% variant of this method
-    {noreply, State};
+handle_method(#'basic.recover_async'{requeue = false}, _, _State) ->
+    rabbit_misc:protocol_error(not_implemented, "requeue=false", []);
 
 handle_method(#'basic.recover'{requeue = Requeue}, Content, State) ->
     {noreply, State2 = #ch{writer_pid = WriterPid}} =
@@ -889,6 +971,11 @@ handle_method(#'queue.purge'{queue = QueueNameBin,
     return_ok(State, NoWait,
               #'queue.purge_ok'{message_count = PurgedMessageCount});
 
+
+handle_method(#'tx.select'{}, _, #ch{confirm_enabled = true}) ->
+    rabbit_misc:protocol_error(
+      precondition_failed, "cannot switch from confirm to tx mode", []);
+
 handle_method(#'tx.select'{}, _, State = #ch{transaction_id = none}) ->
     {reply, #'tx.select_ok'{}, new_tx(State)};
 
@@ -909,6 +996,25 @@ handle_method(#'tx.rollback'{}, _, #ch{transaction_id = none}) ->
 handle_method(#'tx.rollback'{}, _, State) ->
     {reply, #'tx.rollback_ok'{}, internal_rollback(State)};
 
+handle_method(#'confirm.select'{}, _, #ch{transaction_id = TxId})
+  when TxId =/= none ->
+    rabbit_misc:protocol_error(
+      precondition_failed, "cannot switch from tx to confirm mode", []);
+
+handle_method(#'confirm.select'{multiple = Multiple, nowait = NoWait},
+              _, State = #ch{confirm_enabled = false}) ->
+    return_ok(State#ch{confirm_enabled = true, confirm_multiple = Multiple},
+              NoWait, #'confirm.select_ok'{});
+
+handle_method(#'confirm.select'{multiple = Multiple, nowait = NoWait},
+              _, State = #ch{confirm_enabled = true,
+                             confirm_multiple = Multiple}) ->
+    return_ok(State, NoWait, #'confirm.select_ok'{});
+
+handle_method(#'confirm.select'{}, _, #ch{confirm_enabled = true}) ->
+    rabbit_misc:protocol_error(
+      precondition_failed, "cannot change confirm_multiple setting", []);
+
 handle_method(#'channel.flow'{active = true}, _,
               State = #ch{limiter_pid = LimiterPid}) ->
     LimiterPid1 = case rabbit_limiter:unblock(LimiterPid) of
@@ -998,6 +1104,10 @@ basic_return(#basic_message{exchange_name = ExchangeName,
                            routing_key = RoutingKey},
            Content).
 
+ack_record(DeliveryTag, ConsumerTag,
+           _MsgStruct = {_QName, QPid, MsgId, _Redelivered, _Msg}) ->
+    {DeliveryTag, ConsumerTag, {QPid, MsgId}}.
+
 collect_acks(Q, 0, true) ->
     {Q, queue:new()};
 collect_acks(Q, DeliveryTag, Multiple) ->
@@ -1070,8 +1180,7 @@ rollback_and_notify(State) ->
 
 fold_per_queue(F, Acc0, UAQ) ->
     D = rabbit_misc:queue_fold(
-          fun ({_DTag, _CTag,
-                {_QName, QPid, MsgId, _Redelivered, _Message}}, D) ->
+          fun ({_DTag, _CTag, {QPid, MsgId}}, D) ->
                   %% dict:append would avoid the lists:reverse in
                   %% handle_message({recover, true}, ...). However, it
                   %% is significantly slower when going beyond a few
@@ -1133,28 +1242,29 @@ is_message_persistent(Content) ->
             IsPersistent
     end.
 
+process_routing_result(unroutable,    _, MsgSeqNo, Message, State) ->
+    ok = basic_return(Message, State#ch.writer_pid, no_route),
+    send_or_enqueue_ack(MsgSeqNo, undefined, State);
+process_routing_result(not_delivered, _, MsgSeqNo, Message, State) ->
+    ok = basic_return(Message, State#ch.writer_pid, no_consumers),
+    send_or_enqueue_ack(MsgSeqNo, undefined, State);
+process_routing_result(routed,       [], MsgSeqNo,       _, State) ->
+    send_or_enqueue_ack(MsgSeqNo, undefined, State);
+process_routing_result(routed,        _, undefined,      _, State) ->
+    State;
+process_routing_result(routed,    QPids, MsgSeqNo,       _,
+                       State = #ch{queues_for_msg = QFM}) ->
+    QFM1 = dict:store(MsgSeqNo, sets:from_list(QPids), QFM),
+    [maybe_monitor(QPid) || QPid <- QPids],
+    State#ch{queues_for_msg = QFM1}.
+
 lock_message(true, MsgStruct, State = #ch{unacked_message_q = UAMQ}) ->
     State#ch{unacked_message_q = queue:in(MsgStruct, UAMQ)};
 lock_message(false, _MsgStruct, State) ->
     State.
 
-internal_deliver(WriterPid, Notify, ConsumerTag, DeliveryTag,
-                 {_QName, QPid, _MsgId, Redelivered,
-                  #basic_message{exchange_name = ExchangeName,
-                                 routing_key = RoutingKey,
-                                 content = Content}}) ->
-    M = #'basic.deliver'{consumer_tag = ConsumerTag,
-                         delivery_tag = DeliveryTag,
-                         redelivered = Redelivered,
-                         exchange = ExchangeName#resource.name,
-                         routing_key = RoutingKey},
-    ok = case Notify of
-             true  -> rabbit_writer:send_command_and_notify(
-                        WriterPid, QPid, self(), M, Content);
-             false -> rabbit_writer:send_command(WriterPid, M, Content)
-         end.
-
-terminate(_State) ->
+terminate(State) ->
+    stop_confirm_timer(State),
     pg_local:leave(rabbit_channels, self()),
     rabbit_event:notify(channel_closed, [{pid, self()}]).
 
@@ -1214,11 +1324,14 @@ update_measures(Type, QX, Inc, Measure) ->
     put({Type, QX},
         orddict:store(Measure, Cur + Inc, Measures)).
 
-internal_emit_stats(State = #ch{stats_timer = StatsTimer}) ->
+internal_emit_stats(State) ->
+    internal_emit_stats(State, []).
+
+internal_emit_stats(State = #ch{stats_timer = StatsTimer}, Extra) ->
     CoarseStats = infos(?STATISTICS_KEYS, State),
     case rabbit_event:stats_level(StatsTimer) of
         coarse ->
-            rabbit_event:notify(channel_stats, CoarseStats);
+            rabbit_event:notify(channel_stats, Extra ++ CoarseStats);
         fine ->
             FineStats =
                 [{channel_queue_stats,
@@ -1228,7 +1341,8 @@ internal_emit_stats(State = #ch{stats_timer = StatsTimer}) ->
                  {channel_queue_exchange_stats,
                   [{QX, Stats} ||
                       {{queue_exchange_stats, QX}, Stats} <- get()]}],
-            rabbit_event:notify(channel_stats, CoarseStats ++ FineStats)
+            rabbit_event:notify(channel_stats,
+                                Extra ++ CoarseStats ++ FineStats)
     end.
 
 erase_queue_stats(QPid) ->
@@ -1236,3 +1350,42 @@ erase_queue_stats(QPid) ->
     erase({queue_stats, QPid}),
     [erase({queue_exchange_stats, QX}) ||
         {{queue_exchange_stats, QX = {QPid0, _}}, _} <- get(), QPid =:= QPid0].
+
+start_confirm_timer(State = #ch{confirm_tref = undefined}) ->
+    {ok, TRef} = timer:apply_after(?FLUSH_MULTIPLE_ACKS_INTERVAL,
+                                   ?MODULE, flush_multiple_acks, [self()]),
+    State#ch{confirm_tref = TRef};
+start_confirm_timer(State) ->
+    State.
+
+stop_confirm_timer(State = #ch{confirm_tref = undefined}) ->
+    State;
+stop_confirm_timer(State = #ch{confirm_tref = TRef}) ->
+    {ok, cancel} = timer:cancel(TRef),
+    State#ch{confirm_tref = undefined}.
+
+flush_multiple(State = #ch{writer_pid    = WriterPid,
+                           held_confirms = Cs}) ->
+    case gb_sets:is_empty(Cs) of
+        true  -> State#ch{confirm_tref = undefined};
+        false -> [First | Rest] = gb_sets:to_list(Cs),
+                 {Mult, Inds} = find_consecutive_sequence(First, Rest),
+                 ok = rabbit_writer:send_command(
+                        WriterPid,
+                        #'basic.ack'{delivery_tag = Mult, multiple = true}),
+                 ok = lists:foldl(
+                        fun(T, ok) -> rabbit_writer:send_command(
+                                        WriterPid,
+                                        #'basic.ack'{delivery_tag = T})
+                        end, ok, Inds),
+                 State#ch{held_confirms = gb_sets:new(),
+                          confirm_tref  = undefined}
+    end.
+
+%% Find longest sequence of consecutive numbers at the beginning.
+find_consecutive_sequence(Last, []) ->
+    {Last, []};
+find_consecutive_sequence(Last, [N | Ns]) when N == (Last + 1) ->
+    find_consecutive_sequence(N, Ns);
+find_consecutive_sequence(Last, Ns) ->
+    {Last, Ns}.
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 6c0a727b..360217a2 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -32,7 +32,7 @@
 -module(rabbit_control).
 -include("rabbit.hrl").
 
--export([start/0, stop/0, action/5]).
+-export([start/0, stop/0, action/5, diagnostics/1]).
 
 -define(RPC_TIMEOUT, infinity).
 
@@ -50,6 +50,7 @@
         (atom(), node(), [string()], [{string(), any()}],
          fun ((string(), [any()]) -> 'ok'))
         -> 'ok').
+-spec(diagnostics/1 :: (node()) -> [{string(), [any()]}]).
 -spec(usage/0 :: () -> no_return()).
 
 -endif.
@@ -94,9 +95,7 @@ start() ->
             halt();
         {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
             print_error("invalid command '~s'",
-                        [lists:flatten(
-                           rabbit_misc:intersperse(
-                             " ", [atom_to_list(Command) | Args]))]),
+                        [string:join([atom_to_list(Command) | Args], " ")]),
             usage();
         {error, Reason} ->
             print_error("~p", [Reason]),
@@ -118,24 +117,28 @@ fmt_stderr(Format, Args) -> rabbit_misc:format_stderr(Format ++ "~n", Args).
 print_error(Format, Args) -> fmt_stderr("Error: " ++ Format, Args).
 
 print_badrpc_diagnostics(Node) ->
-    fmt_stderr("diagnostics:", []),
+    [fmt_stderr(Fmt, Args) || {Fmt, Args} <- diagnostics(Node)].
+
+diagnostics(Node) ->
     {_NodeName, NodeHost} = rabbit_misc:nodeparts(Node),
-    case net_adm:names(NodeHost) of
-        {error, EpmdReason} ->
-            fmt_stderr("- unable to connect to epmd on ~s: ~w",
-                       [NodeHost, EpmdReason]);
-        {ok, NamePorts} ->
-            fmt_stderr("- nodes and their ports on ~s: ~p",
-                       [NodeHost, [{list_to_atom(Name), Port} ||
-                                      {Name, Port} <- NamePorts]])
-            end,
-    fmt_stderr("- current node: ~w", [node()]),
-    case init:get_argument(home) of
-        {ok, [[Home]]} -> fmt_stderr("- current node home dir: ~s", [Home]);
-        Other          -> fmt_stderr("- no current node home dir: ~p", [Other])
-    end,
-    fmt_stderr("- current node cookie hash: ~s", [rabbit_misc:cookie_hash()]),
-    ok.
+    [
+        {"diagnostics:", []},
+        case net_adm:names(NodeHost) of
+            {error, EpmdReason} ->
+                {"- unable to connect to epmd on ~s: ~w",
+                    [NodeHost, EpmdReason]};
+            {ok, NamePorts} ->
+                {"- nodes and their ports on ~s: ~p",
+                              [NodeHost, [{list_to_atom(Name), Port} ||
+                                          {Name, Port} <- NamePorts]]}
+        end,
+        {"- current node: ~w", [node()]},
+        case init:get_argument(home) of
+            {ok, [[Home]]} -> {"- current node home dir: ~s", [Home]};
+            Other          -> {"- no current node home dir: ~p", [Other]}
+        end,
+        {"- current node cookie hash: ~s", [rabbit_misc:cookie_hash()]}
+    ].
 
 stop() ->
     ok.
@@ -321,7 +324,7 @@ display_info_list(Other, _) ->
     Other.
 
 display_row(Row) ->
-    io:fwrite(lists:flatten(rabbit_misc:intersperse("\t", Row))),
+    io:fwrite(string:join(Row, "\t")),
     io:nl().
 
 -define(IS_U8(X),  (X >= 0 andalso X =< 255)).
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 3a7eaaad..01e962d5 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -180,7 +180,7 @@ assert_equivalence(#exchange{ name = Name },
                    _Type, _Durable, _Internal, _AutoDelete,
                    _Args) ->
     rabbit_misc:protocol_error(
-      not_allowed,
+      precondition_failed,
       "cannot redeclare ~s with different type, durable, " ++
           "internal or autodelete value",
       [rabbit_misc:rs(Name)]).
@@ -234,7 +234,7 @@ info_all(VHostPath, Items) -> map(VHostPath, fun (X) -> info(X, Items) end).
 
 publish(X = #exchange{name = XName}, Delivery) ->
     rabbit_router:deliver(
-      route(Delivery, {queue:from_list([X]), sets:from_list([XName]), []}),
+      route(Delivery, {queue:from_list([X]), XName, []}),
       Delivery).
 
 route(Delivery, {WorkList, SeenXs, QNames}) ->
@@ -258,13 +258,22 @@ process_alternate(_X, Results) ->
     Results.
 
 process_route(#resource{kind = exchange} = XName,
+              {_WorkList, XName, _QNames} = Acc) ->
+    Acc;
+process_route(#resource{kind = exchange} = XName,
+              {WorkList, #resource{kind = exchange} = SeenX, QNames}) ->
+    {case lookup(XName) of
+         {ok, X}            -> queue:in(X, WorkList);
+         {error, not_found} -> WorkList
+     end, gb_sets:from_list([SeenX, XName]), QNames};
+process_route(#resource{kind = exchange} = XName,
               {WorkList, SeenXs, QNames} = Acc) ->
-    case sets:is_element(XName, SeenXs) of
+    case gb_sets:is_element(XName, SeenXs) of
         true  -> Acc;
         false -> {case lookup(XName) of
                       {ok, X}            -> queue:in(X, WorkList);
                       {error, not_found} -> WorkList
-                  end, sets:add_element(XName, SeenXs), QNames}
+                  end, gb_sets:add_element(XName, SeenXs), QNames}
     end;
 process_route(#resource{kind = queue} = QName,
               {WorkList, SeenXs, QNames}) ->
diff --git a/src/rabbit_invariable_queue.erl b/src/rabbit_invariable_queue.erl
deleted file mode 100644
index 5a0532ea..00000000
--- a/src/rabbit_invariable_queue.erl
+++ /dev/null
@@ -1,314 +0,0 @@
-%%   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 Developers of the Original Code are LShift Ltd,
-%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
-%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
-%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
-%%   Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
-%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
-%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
-%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
-%%   (C) 2007-2010 Rabbit Technologies Ltd.
-%%
-%%   All Rights Reserved.
-%%
-%%   Contributor(s): ______________________________________.
-%%
-
--module(rabbit_invariable_queue).
-
--export([init/3, terminate/1, delete_and_terminate/1, purge/1, publish/3,
-         publish_delivered/4, fetch/2, ack/2, tx_publish/4, tx_ack/3,
-         dropwhile/2, tx_rollback/2, tx_commit/4, requeue/3, len/1, is_empty/1,
-         set_ram_duration_target/2, ram_duration/1, needs_idle_timeout/1,
-         idle_timeout/1, handle_pre_hibernate/1, status/1]).
-
--export([start/1, stop/0]).
-
--behaviour(rabbit_backing_queue).
-
--include("rabbit.hrl").
-
--record(iv_state, { queue, qname, durable, len, pending_ack }).
--record(tx, { pending_messages, pending_acks, is_persistent }).
-
--ifdef(use_specs).
-
--type(ack() :: rabbit_guid:guid() | 'blank_ack').
--type(state() :: #iv_state { queue       :: queue(),
-                             qname       :: rabbit_amqqueue:name(),
-                             len         :: non_neg_integer(),
-                             pending_ack :: dict()
-                           }).
--include("rabbit_backing_queue_spec.hrl").
-
--endif.
-
-start(DurableQueues) ->
-    ok = rabbit_sup:start_child(rabbit_persister, [DurableQueues]).
-
-stop() ->
-    ok = rabbit_sup:stop_child(rabbit_persister).
-
-init(QName, IsDurable, Recover) ->
-    Q = queue:from_list(case IsDurable andalso Recover of
-                            true  -> rabbit_persister:queue_content(QName);
-                            false -> []
-                        end),
-    #iv_state { queue       = Q,
-                qname       = QName,
-                durable     = IsDurable,
-                len         = queue:len(Q),
-                pending_ack = dict:new() }.
-
-terminate(State) ->
-    State #iv_state { queue = queue:new(), len = 0, pending_ack = dict:new() }.
-
-delete_and_terminate(State = #iv_state { qname = QName, durable = IsDurable,
-                                         pending_ack = PA }) ->
-    ok = persist_acks(QName, IsDurable, none, dict:fetch_keys(PA), PA),
-    {_PLen, State1} = purge(State),
-    terminate(State1).
-
-purge(State = #iv_state { queue = Q, qname = QName, durable = IsDurable,
-                          len = Len }) ->
-    %% We do not purge messages pending acks.
-    {AckTags, PA} =
-        rabbit_misc:queue_fold(
-          fun ({#basic_message { is_persistent = false },
-                _MsgProps, _IsDelivered}, Acc) ->
-                  Acc;
-              ({Msg = #basic_message { guid = Guid }, MsgProps, IsDelivered},
-               {AckTagsN, PAN}) ->
-                  ok = persist_delivery(QName, IsDurable, IsDelivered, Msg),
-                  {[Guid | AckTagsN], store_ack(Msg, MsgProps, PAN)}
-          end, {[], dict:new()}, Q),
-    ok = persist_acks(QName, IsDurable, none, AckTags, PA),
-    {Len, State #iv_state { len = 0, queue = queue:new() }}.
-
-publish(Msg, MsgProps, State = #iv_state { queue   = Q,
-                                           qname   = QName,
-                                           durable = IsDurable,
-                                           len     = Len }) ->
-    ok = persist_message(QName, IsDurable, none, Msg, MsgProps),
-    State #iv_state { queue = enqueue(Msg, MsgProps, false, Q), len = Len + 1 }.
-
-publish_delivered(false, _Msg, _MsgProps, State) ->
-    {blank_ack, State};
-publish_delivered(true, Msg = #basic_message { guid = Guid },
-                  MsgProps,
-                  State = #iv_state { qname = QName, durable = IsDurable,
-                                      len = 0, pending_ack = PA }) ->
-    ok = persist_message(QName, IsDurable, none, Msg, MsgProps),
-    ok = persist_delivery(QName, IsDurable, false, Msg),
-    {Guid, State #iv_state { pending_ack = store_ack(Msg, MsgProps, PA) }}.
-
-dropwhile(_Pred, State = #iv_state { len = 0 }) ->
-    State;
-dropwhile(Pred, State = #iv_state { queue = Q }) ->
-    {{value, {Msg, MsgProps, IsDelivered}}, Q1} = queue:out(Q),
-    case Pred(MsgProps) of
-        true  -> {_, State1} = fetch_internal(false, Q1, Msg, MsgProps,
-                                              IsDelivered, State),
-                 dropwhile(Pred, State1);
-        false -> State
-     end.
-
-fetch(_AckRequired, State = #iv_state { len = 0 }) ->
-    {empty, State};
-fetch(AckRequired, State = #iv_state { queue = Q }) ->
-    {{value, {Msg, MsgProps, IsDelivered}}, Q1} = queue:out(Q),
-    fetch_internal(AckRequired, Q1, Msg, MsgProps, IsDelivered, State).
-
-fetch_internal(AckRequired, Q1,
-               Msg = #basic_message { guid = Guid },
-               MsgProps, IsDelivered,
-               State = #iv_state { len         = Len,
-                                   qname       = QName,
-                                   durable     = IsDurable,
-                                   pending_ack = PA }) ->
-    Len1 = Len - 1,
-    ok = persist_delivery(QName, IsDurable, IsDelivered, Msg),
-    PA1 = store_ack(Msg, MsgProps, PA),
-    {AckTag, PA2} = case AckRequired of
-                        true  -> {Guid, PA1};
-                        false -> ok = persist_acks(QName, IsDurable, none,
-                                                   [Guid], PA1),
-                                 {blank_ack, PA}
-                    end,
-    {{Msg, IsDelivered, AckTag, Len1},
-     State #iv_state { queue = Q1, len = Len1, pending_ack = PA2 }}.
-
-ack(AckTags, State = #iv_state { qname = QName, durable = IsDurable,
-                                 pending_ack = PA }) ->
-    ok = persist_acks(QName, IsDurable, none, AckTags, PA),
-    PA1 = remove_acks(AckTags, PA),
-    State #iv_state { pending_ack = PA1 }.
-
-tx_publish(Txn, Msg, MsgProps, State = #iv_state { qname   = QName,
-                                                   durable = IsDurable }) ->
-    Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn),
-    store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProps} | Pubs] }),
-    ok = persist_message(QName, IsDurable, Txn, Msg, MsgProps),
-    State.
-
-tx_ack(Txn, AckTags, State = #iv_state { qname = QName, durable = IsDurable,
-                                         pending_ack = PA }) ->
-    Tx = #tx { pending_acks = Acks } = lookup_tx(Txn),
-    store_tx(Txn, Tx #tx { pending_acks = [AckTags | Acks] }),
-    ok = persist_acks(QName, IsDurable, Txn, AckTags, PA),
-    State.
-
-tx_rollback(Txn, State = #iv_state { qname = QName }) ->
-    #tx { pending_acks = AckTags } = lookup_tx(Txn),
-    ok = do_if_persistent(fun rabbit_persister:rollback_transaction/1,
-                          Txn, QName),
-    erase_tx(Txn),
-    {lists:flatten(AckTags), State}.
-
-tx_commit(Txn, Fun, MsgPropsFun, State = #iv_state { qname       = QName,
-                                                     pending_ack = PA,
-                                                     queue       = Q,
-                                                     len         = Len }) ->
-    #tx { pending_acks = AckTags, pending_messages = PubsRev } = lookup_tx(Txn),
-    ok = do_if_persistent(fun rabbit_persister:commit_transaction/1,
-                          Txn, QName),
-    erase_tx(Txn),
-    Fun(),
-    AckTags1 = lists:flatten(AckTags),
-    PA1 = remove_acks(AckTags1, PA),
-    {Q1, Len1} = lists:foldr(fun ({Msg, MsgProps}, {QN, LenN}) ->
-                                     {enqueue(Msg, MsgPropsFun(MsgProps),
-                                              false, QN),
-                                      LenN + 1}
-                             end, {Q, Len}, PubsRev),
-    {AckTags1, State #iv_state { pending_ack = PA1, queue = Q1, len = Len1 }}.
-
-requeue(AckTags, MsgPropsFun, State = #iv_state { pending_ack = PA,
-                                                  queue       = Q,
-                                                  len         = Len }) ->
-    %% We don't need to touch the persister here - the persister will
-    %% already have these messages published and delivered as
-    %% necessary. The complication is that the persister's seq_id will
-    %% now be wrong, given the position of these messages in our queue
-    %% here. However, the persister's seq_id is only used for sorting
-    %% on startup, and requeue is silent as to where the requeued
-    %% messages should appear, thus the persister is permitted to sort
-    %% based on seq_id, even though it'll likely give a different
-    %% order to the last known state of our queue, prior to shutdown.
-    {Q1, Len1} = lists:foldl(
-                   fun (Guid, {QN, LenN}) ->
-                           {Msg = #basic_message {}, MsgProps}
-                               = dict:fetch(Guid, PA),
-                           {enqueue(Msg, MsgPropsFun(MsgProps), true, QN),
-                            LenN + 1}
-                   end, {Q, Len}, AckTags),
-    PA1 = remove_acks(AckTags, PA),
-    State #iv_state { pending_ack = PA1, queue = Q1, len = Len1 }.
-
-enqueue(Msg, MsgProps, IsDelivered, Q) ->
-    queue:in({Msg, MsgProps, IsDelivered}, Q).
-
-len(#iv_state { len = Len }) -> Len.
-
-is_empty(State) -> 0 == len(State).
-
-set_ram_duration_target(_DurationTarget, State) -> State.
-
-ram_duration(State) -> {0, State}.
-
-needs_idle_timeout(_State) -> false.
-
-idle_timeout(State) -> State.
-
-handle_pre_hibernate(State) -> State.
-
-status(_State) -> [].
-
-%%----------------------------------------------------------------------------
-
-remove_acks(AckTags, PA) -> lists:foldl(fun dict:erase/2, PA, AckTags).
-
-store_ack(Msg = #basic_message { guid = Guid }, MsgProps, PA) ->
-    dict:store(Guid, {Msg, MsgProps}, PA).
-
-%%----------------------------------------------------------------------------
-
-lookup_tx(Txn) ->
-    case get({txn, Txn}) of
-        undefined -> #tx { pending_messages = [],
-                           pending_acks     = [],
-                           is_persistent    = false };
-        V         -> V
-    end.
-
-store_tx(Txn, Tx) ->
-    put({txn, Txn}, Tx).
-
-erase_tx(Txn) ->
-    erase({txn, Txn}).
-
-mark_tx_persistent(Txn) ->
-    store_tx(Txn, (lookup_tx(Txn)) #tx { is_persistent = true }).
-
-is_tx_persistent(Txn) ->
-    (lookup_tx(Txn)) #tx.is_persistent.
-
-do_if_persistent(F, Txn, QName) ->
-    ok = case is_tx_persistent(Txn) of
-             false -> ok;
-             true  -> F({Txn, QName})
-         end.
-
-%%----------------------------------------------------------------------------
-
-persist_message(QName, true, Txn, Msg = #basic_message {
-                                    is_persistent = true }, MsgProps) ->
-    Msg1 = Msg #basic_message {
-             %% don't persist any recoverable decoded properties
-             content = rabbit_binary_parser:clear_decoded_content(
-                         Msg #basic_message.content)},
-    persist_work(Txn, QName,
-                 [{publish, Msg1, MsgProps,
-                   {QName, Msg1 #basic_message.guid}}]);
-persist_message(_QName, _IsDurable, _Txn, _Msg, _MsgProps) ->
-    ok.
-
-persist_delivery(QName, true, false, #basic_message { is_persistent = true,
-                                                      guid = Guid }) ->
-    persist_work(none, QName, [{deliver, {QName, Guid}}]);
-persist_delivery(_QName, _IsDurable, _IsDelivered, _Msg) ->
-    ok.
-
-persist_acks(QName, true, Txn, AckTags, PA) ->
-    persist_work(Txn, QName,
-                 [{ack, {QName, Guid}} || Guid <- AckTags,
-                                          begin
-                                              {Msg, _MsgProps}
-                                                  = dict:fetch(Guid, PA),
-                                              Msg #basic_message.is_persistent
-                                          end]);
-persist_acks(_QName, _IsDurable, _Txn, _AckTags, _PA) ->
-    ok.
-
-persist_work(_Txn,_QName, []) ->
-    ok;
-persist_work(none, _QName, WorkList) ->
-    rabbit_persister:dirty_work(WorkList);
-persist_work(Txn, QName, WorkList) ->
-    mark_tx_persistent(Txn),
-    rabbit_persister:extend_transaction({Txn, QName}, WorkList).
diff --git a/src/rabbit_log.erl b/src/rabbit_log.erl
index 863f77e7..a1a8364c 100644
--- a/src/rabbit_log.erl
+++ b/src/rabbit_log.erl
@@ -41,9 +41,6 @@
 -export([debug/1, debug/2, message/4, info/1, info/2,
          warning/1, warning/2, error/1, error/2]).
 
--import(io).
--import(error_logger).
-
 -define(SERVER, ?MODULE).
 
 %%----------------------------------------------------------------------------
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 0522afdc..52d76ac4 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -50,7 +50,7 @@
 -export([execute_mnesia_transaction/1]).
 -export([ensure_ok/2]).
 -export([makenode/1, nodeparts/1, cookie_hash/0, tcp_name/3]).
--export([intersperse/2, upmap/2, map_in_order/2]).
+-export([upmap/2, map_in_order/2]).
 -export([table_fold/3]).
 -export([dirty_read_all/1, dirty_foreach_key/2, dirty_dump_log/1]).
 -export([read_term_file/1, write_term_file/2]).
@@ -61,16 +61,12 @@
 -export([sort_field_table/1]).
 -export([pid_to_string/1, string_to_pid/1]).
 -export([version_compare/2, version_compare/3]).
--export([recursive_delete/1, dict_cons/3, orddict_cons/3,
+-export([recursive_delete/1, recursive_copy/2, dict_cons/3, orddict_cons/3,
          unlink_and_capture_exit/1]).
 -export([get_options/2]).
--export([all_module_attributes/1, build_acyclic_graph/4]).
+-export([all_module_attributes/1, build_acyclic_graph/3]).
 -export([now_ms/0]).
-
--import(mnesia).
--import(lists).
--import(cover).
--import(disk_log).
+-export([lock_file/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -86,10 +82,9 @@
       :: rabbit_types:channel_exit() | rabbit_types:connection_exit()).
 -type(digraph_label() :: term()).
 -type(graph_vertex_fun() ::
-        fun ((atom(), [term()]) -> {digraph:vertex(), digraph_label()})).
+        fun ((atom(), [term()]) -> [{digraph:vertex(), digraph_label()}])).
 -type(graph_edge_fun() ::
-        fun ((atom(), [term()]) -> {digraph:vertex(), digraph:vertex()})).
--type(graph_error_fun() :: fun ((any()) -> any() | no_return())).
+        fun ((atom(), [term()]) -> [{digraph:vertex(), digraph:vertex()}])).
 
 -spec(method_record_type/1 :: (rabbit_framing:amqp_method_record())
                               -> rabbit_framing:amqp_method_name()).
@@ -155,7 +150,6 @@
 -spec(tcp_name/3 ::
         (atom(), inet:ip_address(), rabbit_networking:ip_port())
         -> atom()).
--spec(intersperse/2 :: (A, [A]) -> [A]).
 -spec(upmap/2 :: (fun ((A) -> B), [A]) -> [B]).
 -spec(map_in_order/2 :: (fun ((A) -> B), [A]) -> [B]).
 -spec(table_fold/3 :: (fun ((any(), A) -> A), A, atom()) -> A).
@@ -185,16 +179,24 @@
 -spec(recursive_delete/1 ::
         ([file:filename()])
         -> rabbit_types:ok_or_error({file:filename(), any()})).
+-spec(recursive_copy/2 ::
+        (file:filename(), file:filename())
+        -> rabbit_types:ok_or_error({file:filename(), file:filename(), any()})).
 -spec(dict_cons/3 :: (any(), any(), dict()) -> dict()).
 -spec(orddict_cons/3 :: (any(), any(), orddict:orddict()) -> orddict:orddict()).
 -spec(unlink_and_capture_exit/1 :: (pid()) -> 'ok').
 -spec(get_options/2 :: ([optdef()], [string()])
                        -> {[string()], [{string(), any()}]}).
 -spec(all_module_attributes/1 :: (atom()) -> [{atom(), [term()]}]).
--spec(build_acyclic_graph/4 :: (graph_vertex_fun(), graph_edge_fun(),
-                                graph_error_fun(), [{atom(), [term()]}]) ->
-                                    digraph()).
+-spec(build_acyclic_graph/3 ::
+        (graph_vertex_fun(), graph_edge_fun(), [{atom(), [term()]}])
+        -> rabbit_types:ok_or_error2(digraph(),
+                                     {'vertex', 'duplicate', digraph:vertex()} |
+                                     {'edge', ({bad_vertex, digraph:vertex()} |
+                                               {bad_edge, [digraph:vertex()]}),
+                                      digraph:vertex(), digraph:vertex()})).
 -spec(now_ms/0 :: () -> non_neg_integer()).
+-spec(lock_file/1 :: (file:filename()) -> rabbit_types:ok_or_error('eexist')).
 
 -endif.
 
@@ -240,7 +242,7 @@ assert_args_equivalence1(Orig, New, Name, Key) ->
     case {table_lookup(Orig, Key), table_lookup(New, Key)} of
         {Same, Same}  -> ok;
         {Orig1, New1} -> protocol_error(
-                           not_allowed,
+                           precondition_failed,
                            "inequivalent arg '~s' for ~s:  "
                            "required ~w, received ~w",
                            [Key, rabbit_misc:rs(Name), New1, Orig1])
@@ -414,10 +416,6 @@ tcp_name(Prefix, IPAddress, Port)
         io_lib:format("~w_~s:~w",
                       [Prefix, inet_parse:ntoa(IPAddress), Port]))).
 
-intersperse(_, []) -> [];
-intersperse(_, [E]) -> [E];
-intersperse(Sep, [E|T]) -> [E, Sep | intersperse(Sep, T)].
-
 %% This is a modified version of Luke Gorrie's pmap -
 %% http://lukego.livejournal.com/6753.html - that doesn't care about
 %% the order in which results are received.
@@ -689,6 +687,33 @@ recursive_delete1(Path) ->
                  end
     end.
 
+recursive_copy(Src, Dest) ->
+    case filelib:is_dir(Src) of
+        false -> case file:copy(Src, Dest) of
+                     {ok, _Bytes}    -> ok;
+                     {error, enoent} -> ok; %% Path doesn't exist anyway
+                     {error, Err}    -> {error, {Src, Dest, Err}}
+                 end;
+        true  -> case file:list_dir(Src) of
+                     {ok, FileNames} ->
+                         case file:make_dir(Dest) of
+                             ok ->
+                                 lists:foldl(
+                                   fun (FileName, ok) ->
+                                           recursive_copy(
+                                             filename:join(Src, FileName),
+                                             filename:join(Dest, FileName));
+                                       (_FileName, Error) ->
+                                           Error
+                                   end, ok, FileNames);
+                             {error, Err} ->
+                                 {error, {Src, Dest, Err}}
+                         end;
+                     {error, Err} ->
+                         {error, {Src, Dest, Err}}
+                 end
+    end.
+
 dict_cons(Key, Value, Dict) ->
     dict:update(Key, fun (List) -> [Value | List] end, [Value], Dict).
 
@@ -765,16 +790,30 @@ all_module_attributes(Name) ->
       end, [], Modules).
 
 
-build_acyclic_graph(VertexFun, EdgeFun, ErrorFun, Graph) ->
+build_acyclic_graph(VertexFun, EdgeFun, Graph) ->
     G = digraph:new([acyclic]),
-    [ case digraph:vertex(G, Vertex) of
-          false -> digraph:add_vertex(G, Vertex, Label);
-          _     -> ErrorFun({vertex, duplicate, Vertex})
-      end || {Module, Atts} <- Graph,
-             {Vertex, Label} <- VertexFun(Module, Atts) ],
-    [ case digraph:add_edge(G, From, To) of
-          {error, E} -> ErrorFun({edge, E, From, To});
-          _          -> ok
-      end || {Module, Atts} <- Graph,
-             {From, To} <- EdgeFun(Module, Atts) ],
-    G.
+    try
+        [case digraph:vertex(G, Vertex) of
+             false -> digraph:add_vertex(G, Vertex, Label);
+             _     -> ok = throw({graph_error, {vertex, duplicate, Vertex}})
+         end || {Module, Atts}  <- Graph,
+                {Vertex, Label} <- VertexFun(Module, Atts)],
+        [case digraph:add_edge(G, From, To) of
+             {error, E} -> throw({graph_error, {edge, E, From, To}});
+             _          -> ok
+         end || {Module, Atts} <- Graph,
+                {From, To}     <- EdgeFun(Module, Atts)],
+        {ok, G}
+    catch {graph_error, Reason} ->
+            true = digraph:delete(G),
+            {error, Reason}
+    end.
+
+%% TODO: When we stop supporting Erlang prior to R14, this should be
+%% replaced with file:open [write, exclusive]
+lock_file(Path) ->
+    case filelib:is_file(Path) of
+        true  -> {error, eexist};
+        false -> {ok, Lock} = file:open(Path, [write]),
+                 ok = file:close(Lock)
+    end.
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 9d172269..a62e7a6f 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -34,7 +34,7 @@
 
 -export([ensure_mnesia_dir/0, dir/0, status/0, init/0, is_db_empty/0,
          cluster/1, force_cluster/1, reset/0, force_reset/0,
-         is_clustered/0, empty_ram_only_tables/0]).
+         is_clustered/0, empty_ram_only_tables/0, copy_db/1]).
 
 -export([table_names/0]).
 
@@ -65,6 +65,7 @@
 -spec(is_clustered/0 :: () -> boolean()).
 -spec(empty_ram_only_tables/0 :: () -> 'ok').
 -spec(create_tables/0 :: () -> 'ok').
+-spec(copy_db/1 :: (file:filename()) ->  rabbit_types:ok_or_error(any())).
 
 -endif.
 
@@ -361,41 +362,38 @@ init_db(ClusterNodes, Force) ->
     case mnesia:change_config(extra_db_nodes, ProperClusterNodes) of
         {ok, Nodes} ->
             case Force of
-                false ->
-                    FailedClusterNodes = ProperClusterNodes -- Nodes,
-                    case FailedClusterNodes of
-                        [] -> ok;
-                        _ ->
-                            throw({error, {failed_to_cluster_with,
-                                           FailedClusterNodes,
-                                           "Mnesia could not connect to some nodes."}})
-                    end;
-                _ -> ok
+                false -> FailedClusterNodes = ProperClusterNodes -- Nodes,
+                         case FailedClusterNodes of
+                             [] -> ok;
+                             _  -> throw({error, {failed_to_cluster_with,
+                                                  FailedClusterNodes,
+                                                  "Mnesia could not connect "
+                                                  "to some nodes."}})
+                         end;
+                true  -> ok
             end,
             case {Nodes, mnesia:system_info(use_dir),
                   mnesia:system_info(db_nodes)} of
                 {[], true, [_]} ->
                     %% True single disc node, attempt upgrade
-                    wait_for_tables(),
+                    ok = wait_for_tables(),
                     case rabbit_upgrade:maybe_upgrade() of
-                        ok ->
-                            schema_ok_or_exit();
-                        version_not_available ->
-                            schema_ok_or_move()
+                        ok                    -> ensure_schema_ok();
+                        version_not_available -> schema_ok_or_move()
                     end;
                 {[], true, _} ->
                     %% "Master" (i.e. without config) disc node in cluster,
                     %% verify schema
-                    wait_for_tables(),
-                    version_ok_or_exit(rabbit_upgrade:read_version()),
-                    schema_ok_or_exit();
+                    ok = wait_for_tables(),
+                    ensure_version_ok(rabbit_upgrade:read_version()),
+                    ensure_schema_ok();
                 {[], false, _} ->
                     %% First RAM node in cluster, start from scratch
                     ok = create_schema();
                 {[AnotherNode|_], _, _} ->
                     %% Subsequent node in cluster, catch up
-                    version_ok_or_exit(rabbit_upgrade:read_version()),
-                    version_ok_or_exit(
+                    ensure_version_ok(rabbit_upgrade:read_version()),
+                    ensure_version_ok(
                       rpc:call(AnotherNode, rabbit_upgrade, read_version, [])),
                     IsDiskNode = ClusterNodes == [] orelse
                         lists:member(node(), ClusterNodes),
@@ -405,14 +403,13 @@ init_db(ClusterNodes, Force) ->
                                                        true  -> disc;
                                                        false -> ram
                                                    end),
-                    schema_ok_or_exit()
+                    ensure_schema_ok()
             end;
         {error, Reason} ->
             %% one reason we may end up here is if we try to join
             %% nodes together that are currently running standalone or
             %% are members of a different cluster
-            throw({error, {unable_to_join_cluster,
-                           ClusterNodes, Reason}})
+            throw({error, {unable_to_join_cluster, ClusterNodes, Reason}})
     end.
 
 schema_ok_or_move() ->
@@ -430,22 +427,19 @@ schema_ok_or_move() ->
             ok = create_schema()
     end.
 
-version_ok_or_exit({ok, DiscVersion}) ->
+ensure_version_ok({ok, DiscVersion}) ->
     case rabbit_upgrade:desired_version() of
-        DiscVersion ->
-            ok;
-        DesiredVersion ->
-            exit({schema_mismatch, DesiredVersion, DiscVersion})
+        DiscVersion    ->  ok;
+        DesiredVersion ->  throw({error, {schema_mismatch,
+                                          DesiredVersion, DiscVersion}})
     end;
-version_ok_or_exit({error, _}) ->
+ensure_version_ok({error, _}) ->
     ok = rabbit_upgrade:write_version().
 
-schema_ok_or_exit() ->
+ensure_schema_ok() ->
     case check_schema_integrity() of
-        ok ->
-            ok;
-        {error, Reason} ->
-            exit({schema_invalid, Reason})
+        ok              -> ok;
+        {error, Reason} -> throw({error, {schema_invalid, Reason}})
     end.
 
 create_schema() ->
@@ -481,6 +475,16 @@ move_db() ->
     rabbit_misc:ensure_ok(mnesia:start(), cannot_start_mnesia),
     ok.
 
+copy_db(Destination) ->
+    mnesia:stop(),
+    case rabbit_misc:recursive_copy(dir(), Destination) of
+        ok ->
+            rabbit_misc:ensure_ok(mnesia:start(), cannot_start_mnesia),
+            ok = wait_for_tables();
+        {error, E} ->
+            {error, E}
+    end.
+
 create_tables() ->
     lists:foreach(fun ({Tab, TabDef}) ->
                           TabDef1 = proplists:delete(match, TabDef),
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index fd84109b..e8b4e8e2 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -34,7 +34,7 @@
 -behaviour(gen_server2).
 
 -export([start_link/4, successfully_recovered_state/1,
-         client_init/2, client_terminate/1, client_delete_and_terminate/1,
+         client_init/3, client_terminate/1, client_delete_and_terminate/1,
          client_ref/1,
          write/3, read/2, contains/2, remove/2, release/2, sync/3]).
 
@@ -83,7 +83,9 @@
           cur_file_cache_ets,     %% tid of current file cache table
           client_refs,            %% set of references of all registered clients
           successfully_recovered, %% boolean: did we recover state?
-          file_size_limit         %% how big are our files allowed to get?
+          file_size_limit,        %% how big are our files allowed to get?
+          client_ondisk_callback, %% client ref to callback function mapping
+          cref_to_guids           %% client ref to synced messages mapping
          }).
 
 -record(client_msstate,
@@ -138,16 +140,18 @@
                       file_handles_ets   :: ets:tid(),
                       file_summary_ets   :: ets:tid(),
                       dedup_cache_ets    :: ets:tid(),
-                      cur_file_cache_ets :: ets:tid() }).
+                      cur_file_cache_ets :: ets:tid()}).
 -type(startup_fun_state() ::
         {(fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A})),
          A}).
+-type(maybe_guid_fun() :: 'undefined' | fun ((gb_set()) -> any())).
 
 -spec(start_link/4 ::
         (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(), client_ref()) -> client_msstate()).
+-spec(client_init/3 :: (server(), client_ref(), maybe_guid_fun()) ->
+             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()).
@@ -334,10 +338,11 @@ start_link(Server, Dir, ClientRefs, StartupFunState) ->
 successfully_recovered_state(Server) ->
     gen_server2:call(Server, successfully_recovered_state, infinity).
 
-client_init(Server, Ref) ->
+client_init(Server, Ref, MsgOnDiskFun) ->
     {IState, IModule, Dir, GCPid,
      FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts} =
-        gen_server2:call(Server, {new_client_state, Ref}, infinity),
+        gen_server2:call(Server, {new_client_state, Ref, MsgOnDiskFun},
+                         infinity),
     #client_msstate { server             = Server,
                       client_ref         = Ref,
                       file_handle_cache  = dict:new(),
@@ -350,9 +355,9 @@ client_init(Server, Ref) ->
                       dedup_cache_ets    = DedupCacheEts,
                       cur_file_cache_ets = CurFileCacheEts }.
 
-client_terminate(CState) ->
+client_terminate(CState = #client_msstate { client_ref = Ref }) ->
     close_all_handles(CState),
-    ok = server_call(CState, client_terminate).
+    ok = server_call(CState, {client_terminate, Ref}).
 
 client_delete_and_terminate(CState = #client_msstate { client_ref = Ref }) ->
     close_all_handles(CState),
@@ -361,9 +366,10 @@ client_delete_and_terminate(CState = #client_msstate { client_ref = Ref }) ->
 client_ref(#client_msstate { client_ref = Ref }) -> Ref.
 
 write(Guid, Msg,
-      CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts }) ->
+      CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts,
+                                 client_ref         = CRef }) ->
     ok = update_msg_cache(CurFileCacheEts, Guid, Msg),
-    ok = server_cast(CState, {write, Guid}).
+    ok = server_cast(CState, {write, CRef, Guid}).
 
 read(Guid,
      CState = #client_msstate { dedup_cache_ets    = DedupCacheEts,
@@ -392,7 +398,8 @@ read(Guid,
 
 contains(Guid, CState) -> server_call(CState, {contains, Guid}).
 remove([],    _CState) -> ok;
-remove(Guids,  CState) -> server_cast(CState, {remove, Guids}).
+remove(Guids, CState = #client_msstate { client_ref = CRef }) ->
+    server_cast(CState, {remove, CRef, Guids}).
 release([],   _CState) -> ok;
 release(Guids, CState) -> server_cast(CState, {release, Guids}).
 sync(Guids, K, CState) -> server_cast(CState, {sync, Guids, K}).
@@ -519,6 +526,13 @@ client_read3(#msg_location { guid = Guid, file = File }, Defer,
             end
     end.
 
+clear_client_callback(CRef,
+                      State = #msstate { client_ondisk_callback = CODC,
+                                         cref_to_guids          = CTG }) ->
+    State #msstate { client_ondisk_callback = dict:erase(CRef, CODC),
+                     cref_to_guids          = dict:erase(CRef, CTG)}.
+
+
 %%----------------------------------------------------------------------------
 %% gen_server callbacks
 %%----------------------------------------------------------------------------
@@ -586,7 +600,9 @@ init([Server, BaseDir, ClientRefs, StartupFunState]) ->
                        cur_file_cache_ets     = CurFileCacheEts,
                        client_refs            = ClientRefs1,
                        successfully_recovered = CleanShutdown,
-                       file_size_limit        = FileSizeLimit
+                       file_size_limit        = FileSizeLimit,
+                       client_ondisk_callback = dict:new(),
+                       cref_to_guids          = dict:new()
                       },
 
     %% If we didn't recover the msg location index then we need to
@@ -615,10 +631,10 @@ init([Server, BaseDir, ClientRefs, StartupFunState]) ->
 
 prioritise_call(Msg, _From, _State) ->
     case Msg of
-        successfully_recovered_state -> 7;
-        {new_client_state, _Ref}     -> 7;
-        {read, _Guid}                -> 2;
-        _                            -> 0
+        successfully_recovered_state    -> 7;
+        {new_client_state, _Ref, _MODC} -> 7;
+        {read, _Guid}                   -> 2;
+        _                               -> 0
     end.
 
 prioritise_cast(Msg, _State) ->
@@ -633,22 +649,29 @@ prioritise_cast(Msg, _State) ->
 handle_call(successfully_recovered_state, _From, State) ->
     reply(State #msstate.successfully_recovered, State);
 
-handle_call({new_client_state, CRef}, _From,
-            State = #msstate { dir                = Dir,
-                               index_state        = IndexState,
-                               index_module       = IndexModule,
-                               file_handles_ets   = FileHandlesEts,
-                               file_summary_ets   = FileSummaryEts,
-                               dedup_cache_ets    = DedupCacheEts,
-                               cur_file_cache_ets = CurFileCacheEts,
-                               client_refs        = ClientRefs,
-                               gc_pid             = GCPid }) ->
+handle_call({new_client_state, CRef, Callback}, _From,
+            State = #msstate { dir                    = Dir,
+                               index_state            = IndexState,
+                               index_module           = IndexModule,
+                               file_handles_ets       = FileHandlesEts,
+                               file_summary_ets       = FileSummaryEts,
+                               dedup_cache_ets        = DedupCacheEts,
+                               cur_file_cache_ets     = CurFileCacheEts,
+                               client_refs            = ClientRefs,
+                               client_ondisk_callback = CODC,
+                               gc_pid                 = GCPid }) ->
+    CODC1 = case Callback of
+                undefined -> CODC;
+                _         -> dict:store(CRef, Callback, CODC)
+            end,
     reply({IndexState, IndexModule, Dir, GCPid,
            FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts},
-          State #msstate { client_refs = sets:add_element(CRef, ClientRefs) });
+          State #msstate { client_refs = sets:add_element(CRef, ClientRefs),
+                           client_ondisk_callback = CODC1 });
 
-handle_call(client_terminate, _From, State) ->
-    reply(ok, State);
+handle_call({client_terminate, CRef}, _From,
+            State) ->
+    reply(ok, clear_client_callback(CRef, State));
 
 handle_call({read, Guid}, From, State) ->
     State1 = read_message(Guid, From, State),
@@ -660,43 +683,63 @@ handle_call({contains, Guid}, From, State) ->
 
 handle_cast({client_delete, CRef},
             State = #msstate { client_refs = ClientRefs }) ->
-    noreply(
-      State #msstate { client_refs = sets:del_element(CRef, ClientRefs) });
+    State1 = clear_client_callback(CRef, State),
+    noreply(State1 #msstate {
+              client_refs = sets:del_element(CRef, ClientRefs) });
+
+handle_cast({write, CRef, Guid},
+            State = #msstate { sum_valid_data         = SumValid,
+                               file_summary_ets       = FileSummaryEts,
+                               current_file           = CurFile,
+                               cur_file_cache_ets     = CurFileCacheEts,
+                               client_ondisk_callback = CODC,
+                               cref_to_guids          = CTG }) ->
 
-handle_cast({write, Guid},
-            State = #msstate { sum_valid_data     = SumValid,
-                               file_summary_ets   = FileSummaryEts,
-                               cur_file_cache_ets = CurFileCacheEts }) ->
     true = 0 =< ets:update_counter(CurFileCacheEts, Guid, {3, -1}),
     [{Guid, Msg, _CacheRefCount}] = ets:lookup(CurFileCacheEts, Guid),
-    case index_lookup(Guid, State) of
+    CTG1 = case dict:find(CRef, CODC) of
+               {ok, _} -> dict:update(CRef, fun(Guids) ->
+                                                    gb_sets:add(Guid, Guids)
+                                            end,
+                                      gb_sets:empty(), CTG);
+               error   -> CTG
+           end,
+    State1 = State #msstate { cref_to_guids = CTG1 },
+    case index_lookup(Guid, State1) of
         not_found ->
-            write_message(Guid, Msg, State);
+            write_message(Guid, Msg, State1);
         #msg_location { ref_count = 0, file = File, total_size = TotalSize } ->
             case ets:lookup(FileSummaryEts, File) of
                 [#file_summary { locked = true }] ->
-                    ok = index_delete(Guid, State),
-                    write_message(Guid, Msg, State);
+                    ok = index_delete(Guid, State1),
+                    write_message(Guid, Msg, State1);
                 [#file_summary {}] ->
-                    ok = index_update_ref_count(Guid, 1, State),
+                    ok = index_update_ref_count(Guid, 1, State1),
                     [_] = ets:update_counter(
                             FileSummaryEts, File,
                             [{#file_summary.valid_total_size, TotalSize}]),
-                    noreply(State #msstate {
+                    noreply(State1 #msstate {
                               sum_valid_data = SumValid + TotalSize })
             end;
-        #msg_location { ref_count = RefCount } ->
+        #msg_location { ref_count = RefCount, file = File } ->
             %% We already know about it, just update counter. Only
             %% update field otherwise bad interaction with concurrent GC
-            ok = index_update_ref_count(Guid, RefCount + 1, State),
-            noreply(State)
+            ok = index_update_ref_count(Guid, RefCount + 1, State1),
+            CTG2 = case {dict:find(CRef, CODC), File} of
+                       {{ok, _},   CurFile} -> CTG1;
+                       {{ok, Fun}, _}       -> Fun(gb_sets:singleton(Guid)),
+                                               CTG;
+                       _                    -> CTG1
+                   end,
+            noreply(State #msstate { cref_to_guids = CTG2 })
     end;
 
-handle_cast({remove, Guids}, State) ->
+handle_cast({remove, CRef, Guids}, State) ->
     State1 = lists:foldl(
                fun (Guid, State2) -> remove_message(Guid, State2) end,
                State, Guids),
-    noreply(maybe_compact(State1));
+    State2 = client_confirm(CRef, gb_sets:from_list(Guids), State1),
+    noreply(maybe_compact(State2));
 
 handle_cast({release, Guids}, State =
                 #msstate { dedup_cache_ets = DedupCacheEts }) ->
@@ -794,14 +837,19 @@ reply(Reply, State) ->
     {State1, Timeout} = next_state(State),
     {reply, Reply, State1, Timeout}.
 
-next_state(State = #msstate { on_sync = [], sync_timer_ref = undefined }) ->
-    {State, hibernate};
-next_state(State = #msstate { sync_timer_ref = undefined }) ->
-    {start_sync_timer(State), 0};
-next_state(State = #msstate { on_sync = [] }) ->
-    {stop_sync_timer(State), hibernate};
-next_state(State) ->
-    {State, 0}.
+next_state(State = #msstate { sync_timer_ref = undefined,
+                              on_sync        = Syncs,
+                              cref_to_guids  = CTG }) ->
+    case {Syncs, dict:size(CTG)} of
+        {[], 0} -> {State, hibernate};
+        _       -> {start_sync_timer(State), 0}
+    end;
+next_state(State = #msstate { on_sync       = Syncs,
+                              cref_to_guids = CTG }) ->
+    case {Syncs, dict:size(CTG)} of
+        {[], 0} -> {stop_sync_timer(State), hibernate};
+        _       -> {State, 0}
+    end.
 
 start_sync_timer(State = #msstate { sync_timer_ref = undefined }) ->
     {ok, TRef} = timer:apply_after(?SYNC_INTERVAL, ?MODULE, sync, [self()]),
@@ -813,15 +861,23 @@ stop_sync_timer(State = #msstate { sync_timer_ref = TRef }) ->
     {ok, cancel} = timer:cancel(TRef),
     State #msstate { sync_timer_ref = undefined }.
 
-internal_sync(State = #msstate { current_file_handle = CurHdl,
-                                 on_sync = Syncs }) ->
+internal_sync(State = #msstate { current_file_handle    = CurHdl,
+                                 on_sync                = Syncs,
+                                 cref_to_guids          = CTG }) ->
     State1 = stop_sync_timer(State),
-    case Syncs of
-        [] -> State1;
-        _  -> ok = file_handle_cache:sync(CurHdl),
-              lists:foreach(fun (K) -> K() end, lists:reverse(Syncs)),
-              State1 #msstate { on_sync = [] }
-    end.
+    CGs = dict:fold(fun (CRef, Guids, NS) ->
+                            case gb_sets:is_empty(Guids) of
+                                true  -> NS;
+                                false -> [{CRef, Guids} | NS]
+                            end
+                    end, [], CTG),
+    if Syncs =:= [] andalso CGs =:= [] -> ok;
+       true                            -> file_handle_cache:sync(CurHdl)
+    end,
+    lists:foreach(fun (K) -> K() end, lists:reverse(Syncs)),
+    [client_confirm(CRef, Guids, State1) || {CRef, Guids} <- CGs],
+    State1 #msstate { cref_to_guids = dict:new(), on_sync = [] }.
+
 
 write_message(Guid, Msg,
               State = #msstate { current_file_handle = CurHdl,
@@ -999,6 +1055,25 @@ orddict_store(Key, Val, Dict) ->
     false = orddict:is_key(Key, Dict),
     orddict:store(Key, Val, Dict).
 
+client_confirm(CRef, Guids,
+               State = #msstate { client_ondisk_callback = CODC,
+                                  cref_to_guids          = CTG }) ->
+    case dict:find(CRef, CODC) of
+        {ok, Fun} -> Fun(Guids),
+                     CTG1 = case dict:find(CRef, CTG) of
+                                {ok, Gs} ->
+                                    Guids1 = gb_sets:difference(Gs, Guids),
+                                    case gb_sets:is_empty(Guids1) of
+                                        true  -> dict:erase(CRef, CTG);
+                                        false -> dict:store(CRef, Guids1, CTG)
+                                    end;
+                                error    -> CTG
+                            end,
+                     State #msstate { cref_to_guids = CTG1 };
+        error -> State
+    end.
+
+
 %%----------------------------------------------------------------------------
 %% file helper functions
 %%----------------------------------------------------------------------------
diff --git a/src/rabbit_multi.erl b/src/rabbit_multi.erl
index 0440dbe4..0030216e 100644
--- a/src/rabbit_multi.erl
+++ b/src/rabbit_multi.erl
@@ -65,8 +65,7 @@ start() ->
                     halt();
                 {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
                     print_error("invalid command '~s'",
-                                [lists:flatten(
-                                   rabbit_misc:intersperse(" ", FullCommand))]),
+                                [string:join(FullCommand, " ")]),
                     usage();
                 timeout ->
                     print_error("timeout starting some nodes.", []),
diff --git a/src/rabbit_net.erl b/src/rabbit_net.erl
index 0940dce2..89954b06 100644
--- a/src/rabbit_net.erl
+++ b/src/rabbit_net.erl
@@ -32,9 +32,9 @@
 -module(rabbit_net).
 -include("rabbit.hrl").
 
--export([async_recv/3, close/1, controlling_process/2,
-        getstat/2, peername/1, peercert/1, port_command/2,
-        send/2, sockname/1, is_ssl/1]).
+-export([is_ssl/1, controlling_process/2, getstat/2,
+         async_recv/3, port_command/2, send/2, close/1,
+         sockname/1, peername/1, peercert/1]).
 
 %%---------------------------------------------------------------------------
 
@@ -49,26 +49,25 @@
 -type(ok_or_any_error() :: rabbit_types:ok_or_error(any())).
 -type(socket() :: port() | #ssl_socket{}).
 
+-spec(is_ssl/1 :: (socket()) -> boolean()).
+-spec(controlling_process/2 :: (socket(), pid()) -> ok_or_any_error()).
+-spec(getstat/2 ::
+        (socket(), [stat_option()])
+        -> ok_val_or_error([{stat_option(), integer()}])).
 -spec(async_recv/3 ::
         (socket(), integer(), timeout()) -> rabbit_types:ok(any())).
--spec(close/1 :: (socket()) -> ok_or_any_error()).
--spec(controlling_process/2 :: (socket(), pid()) -> ok_or_any_error()).
 -spec(port_command/2 :: (socket(), iolist()) -> 'true').
--spec(send/2 ::
-        (socket(), binary() | iolist()) -> ok_or_any_error()).
+-spec(send/2 :: (socket(), binary() | iolist()) -> ok_or_any_error()).
+-spec(close/1 :: (socket()) -> ok_or_any_error()).
+-spec(sockname/1 ::
+        (socket())
+        -> ok_val_or_error({inet:ip_address(), rabbit_networking:ip_port()})).
 -spec(peername/1 ::
         (socket())
         -> ok_val_or_error({inet:ip_address(), rabbit_networking:ip_port()})).
 -spec(peercert/1 ::
         (socket())
         -> 'nossl' | ok_val_or_error(rabbit_ssl:certificate())).
--spec(sockname/1 ::
-        (socket())
-        -> ok_val_or_error({inet:ip_address(), rabbit_networking:ip_port()})).
--spec(is_ssl/1 :: (socket()) -> boolean()).
--spec(getstat/2 ::
-        (socket(), [stat_option()])
-        -> ok_val_or_error([{stat_option(), integer()}])).
 
 -endif.
 
@@ -76,6 +75,18 @@
 
 -define(IS_SSL(Sock), is_record(Sock, ssl_socket)).
 
+is_ssl(Sock) -> ?IS_SSL(Sock).
+
+controlling_process(Sock, Pid) when ?IS_SSL(Sock) ->
+    ssl:controlling_process(Sock#ssl_socket.ssl, Pid);
+controlling_process(Sock, Pid) when is_port(Sock) ->
+    gen_tcp:controlling_process(Sock, Pid).
+
+getstat(Sock, Stats) when ?IS_SSL(Sock) ->
+    inet:getstat(Sock#ssl_socket.tcp, Stats);
+getstat(Sock, Stats) when is_port(Sock) ->
+    inet:getstat(Sock, Stats).
+
 async_recv(Sock, Length, Timeout) when ?IS_SSL(Sock) ->
     Pid = self(),
     Ref = make_ref(),
@@ -90,31 +101,6 @@ async_recv(Sock, Length, infinity) when is_port(Sock) ->
 async_recv(Sock, Length, Timeout) when is_port(Sock) ->
     prim_inet:async_recv(Sock, Length, Timeout).
 
-close(Sock) when ?IS_SSL(Sock) ->
-    ssl:close(Sock#ssl_socket.ssl);
-close(Sock) when is_port(Sock) ->
-    gen_tcp:close(Sock).
-
-controlling_process(Sock, Pid) when ?IS_SSL(Sock) ->
-    ssl:controlling_process(Sock#ssl_socket.ssl, Pid);
-controlling_process(Sock, Pid) when is_port(Sock) ->
-    gen_tcp:controlling_process(Sock, Pid).
-
-getstat(Sock, Stats) when ?IS_SSL(Sock) ->
-    inet:getstat(Sock#ssl_socket.tcp, Stats);
-getstat(Sock, Stats) when is_port(Sock) ->
-    inet:getstat(Sock, Stats).
-
-peername(Sock) when ?IS_SSL(Sock) ->
-    ssl:peername(Sock#ssl_socket.ssl);
-peername(Sock) when is_port(Sock) ->
-    inet:peername(Sock).
-
-peercert(Sock) when ?IS_SSL(Sock) ->
-    ssl:peercert(Sock#ssl_socket.ssl);
-peercert(Sock) when is_port(Sock) ->
-    nossl.
-
 port_command(Sock, Data) when ?IS_SSL(Sock) ->
     case ssl:send(Sock#ssl_socket.ssl, Data) of
         ok              -> self() ! {inet_reply, Sock, ok},
@@ -124,16 +110,17 @@ port_command(Sock, Data) when ?IS_SSL(Sock) ->
 port_command(Sock, Data) when is_port(Sock) ->
     erlang:port_command(Sock, Data).
 
-send(Sock, Data) when ?IS_SSL(Sock) ->
-    ssl:send(Sock#ssl_socket.ssl, Data);
-send(Sock, Data) when is_port(Sock) ->
-    gen_tcp:send(Sock, Data).
+send(Sock, Data) when ?IS_SSL(Sock) -> ssl:send(Sock#ssl_socket.ssl, Data);
+send(Sock, Data) when is_port(Sock) -> gen_tcp:send(Sock, Data).
+
+close(Sock)      when ?IS_SSL(Sock) -> ssl:close(Sock#ssl_socket.ssl);
+close(Sock)      when is_port(Sock) -> gen_tcp:close(Sock).
 
+sockname(Sock)   when ?IS_SSL(Sock) -> ssl:sockname(Sock#ssl_socket.ssl);
+sockname(Sock)   when is_port(Sock) -> inet:sockname(Sock).
 
-sockname(Sock) when ?IS_SSL(Sock) ->
-    ssl:sockname(Sock#ssl_socket.ssl);
-sockname(Sock) when is_port(Sock) ->
-    inet:sockname(Sock).
+peername(Sock)   when ?IS_SSL(Sock) -> ssl:peername(Sock#ssl_socket.ssl);
+peername(Sock)   when is_port(Sock) -> inet:peername(Sock).
 
-is_ssl(Sock) ->
-    ?IS_SSL(Sock).
+peercert(Sock)   when ?IS_SSL(Sock) -> ssl:peercert(Sock#ssl_socket.ssl);
+peercert(Sock)   when is_port(Sock) -> nossl.
diff --git a/src/rabbit_persister.erl b/src/rabbit_persister.erl
deleted file mode 100644
index 11056c8e..00000000
--- a/src/rabbit_persister.erl
+++ /dev/null
@@ -1,496 +0,0 @@
-%%   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 Developers of the Original Code are LShift Ltd,
-%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
-%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
-%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
-%%   Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
-%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
-%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
-%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
-%%   (C) 2007-2010 Rabbit Technologies Ltd.
-%%
-%%   All Rights Reserved.
-%%
-%%   Contributor(s): ______________________________________.
-%%
-
--module(rabbit_persister).
-
--behaviour(gen_server).
-
--export([start_link/1]).
-
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
-         terminate/2, code_change/3]).
-
--export([transaction/1, extend_transaction/2, dirty_work/1,
-         commit_transaction/1, rollback_transaction/1,
-         force_snapshot/0, queue_content/1]).
-
--include("rabbit.hrl").
-
--define(SERVER, ?MODULE).
-
--define(LOG_BUNDLE_DELAY, 5).
--define(COMPLETE_BUNDLE_DELAY, 2).
-
--define(PERSISTER_LOG_FORMAT_VERSION, {2, 6}).
-
--record(pstate, {log_handle, entry_count, deadline,
-                 pending_logs, pending_replies, snapshot}).
-
-%% two tables for efficient persistency
-%% one maps a key to a message
-%% the other maps a key to one or more queues.
-%% The aim is to reduce the overload of storing a message multiple times
-%% when it appears in several queues.
--record(psnapshot, {transactions, messages, queues, next_seq_id}).
-
-%%----------------------------------------------------------------------------
-
--ifdef(use_specs).
-
--type(pkey() :: rabbit_guid:guid()).
--type(pmsg() :: {rabbit_amqqueue:name(), pkey()}).
-
--type(work_item() ::
-      {publish,
-       rabbit_types:message(), rabbit_types:message_properties(), pmsg()} |
-      {deliver, pmsg()} |
-      {ack, pmsg()}).
-
--spec(start_link/1 :: ([rabbit_amqqueue:name()]) ->
-                           rabbit_types:ok_pid_or_error()).
--spec(transaction/1 :: ([work_item()]) -> 'ok').
--spec(extend_transaction/2 ::
-        ({rabbit_types:txn(), rabbit_amqqueue:name()}, [work_item()])
-        -> 'ok').
--spec(dirty_work/1 :: ([work_item()]) -> 'ok').
--spec(commit_transaction/1 ::
-        ({rabbit_types:txn(), rabbit_amqqueue:name()}) -> 'ok').
--spec(rollback_transaction/1 ::
-        ({rabbit_types:txn(), rabbit_amqqueue:name()}) -> 'ok').
--spec(force_snapshot/0 :: () -> 'ok').
--spec(queue_content/1 ::
-        (rabbit_amqqueue:name()) -> [{rabbit_types:message(), boolean()}]).
-
--endif.
-
-%%----------------------------------------------------------------------------
-
-start_link(DurableQueues) ->
-    gen_server:start_link({local, ?SERVER}, ?MODULE, [DurableQueues], []).
-
-transaction(MessageList) ->
-    ?LOGDEBUG("transaction ~p~n", [MessageList]),
-    TxnKey = rabbit_guid:guid(),
-    gen_server:call(?SERVER, {transaction, TxnKey, MessageList}, infinity).
-
-extend_transaction(TxnKey, MessageList) ->
-    ?LOGDEBUG("extend_transaction ~p ~p~n", [TxnKey, MessageList]),
-    gen_server:cast(?SERVER, {extend_transaction, TxnKey, MessageList}).
-
-dirty_work(MessageList) ->
-    ?LOGDEBUG("dirty_work ~p~n", [MessageList]),
-    gen_server:cast(?SERVER, {dirty_work, MessageList}).
-
-commit_transaction(TxnKey) ->
-    ?LOGDEBUG("commit_transaction ~p~n", [TxnKey]),
-    gen_server:call(?SERVER, {commit_transaction, TxnKey}, infinity).
-
-rollback_transaction(TxnKey) ->
-    ?LOGDEBUG("rollback_transaction ~p~n", [TxnKey]),
-    gen_server:cast(?SERVER, {rollback_transaction, TxnKey}).
-
-force_snapshot() ->
-    gen_server:call(?SERVER, force_snapshot, infinity).
-
-queue_content(QName) ->
-    gen_server:call(?SERVER, {queue_content, QName}, infinity).
-
-%%--------------------------------------------------------------------
-
-init([DurableQueues]) ->
-    process_flag(trap_exit, true),
-    FileName = base_filename(),
-    ok = filelib:ensure_dir(FileName),
-    Snapshot = #psnapshot{transactions = dict:new(),
-                          messages     = ets:new(messages, []),
-                          queues       = ets:new(queues, [ordered_set]),
-                          next_seq_id  = 0},
-    LogHandle =
-        case disk_log:open([{name, rabbit_persister},
-                            {head, current_snapshot(Snapshot)},
-                            {file, FileName}]) of
-            {ok, LH} -> LH;
-            {repaired, LH, {recovered, Recovered}, {badbytes, Bad}} ->
-                WarningFun = if
-                                 Bad > 0 -> fun rabbit_log:warning/2;
-                                 true    -> fun rabbit_log:info/2
-                             end,
-                WarningFun("Repaired persister log - ~p recovered, ~p bad~n",
-                           [Recovered, Bad]),
-                LH
-        end,
-    {Res, NewSnapshot} =
-        internal_load_snapshot(LogHandle, DurableQueues, Snapshot),
-    case Res of
-        ok ->
-            ok = take_snapshot(LogHandle, NewSnapshot);
-        {error, Reason} ->
-            rabbit_log:error("Failed to load persister log: ~p~n", [Reason]),
-            ok = take_snapshot_and_save_old(LogHandle, NewSnapshot)
-    end,
-    State = #pstate{log_handle        = LogHandle,
-                    entry_count       = 0,
-                    deadline          = infinity,
-                    pending_logs      = [],
-                    pending_replies   = [],
-                    snapshot          = NewSnapshot},
-    {ok, State}.
-
-handle_call({transaction, Key, MessageList}, From, State) ->
-    NewState = internal_extend(Key, MessageList, State),
-    do_noreply(internal_commit(From, Key, NewState));
-handle_call({commit_transaction, TxnKey}, From, State) ->
-    do_noreply(internal_commit(From, TxnKey, State));
-handle_call(force_snapshot, _From, State) ->
-    do_reply(ok, flush(true, State));
-handle_call({queue_content, QName}, _From,
-            State = #pstate{snapshot = #psnapshot{messages = Messages,
-                                                  queues   = Queues}}) ->
-    MatchSpec= [{{{QName,'$1'}, '$2', '$3', '$4'}, [], 
-                 [{{'$4', '$1', '$2', '$3'}}]}],
-    do_reply([{ets:lookup_element(Messages, K, 2), MP, D} ||
-                 {_, K, D, MP} <- lists:sort(ets:select(Queues, MatchSpec))],
-             State);
-handle_call(_Request, _From, State) ->
-    {noreply, State}.
-
-handle_cast({rollback_transaction, TxnKey}, State) ->
-    do_noreply(internal_rollback(TxnKey, State));
-handle_cast({dirty_work, MessageList}, State) ->
-    do_noreply(internal_dirty_work(MessageList, State));
-handle_cast({extend_transaction, TxnKey, MessageList}, State) ->
-    do_noreply(internal_extend(TxnKey, MessageList, State));
-handle_cast(_Msg, State) ->
-    {noreply, State}.
-
-handle_info(timeout, State = #pstate{deadline = infinity}) ->
-    State1 = flush(true, State),
-    {noreply, State1, hibernate};
-handle_info(timeout, State) ->
-    do_noreply(flush(State));
-handle_info(_Info, State) ->
-    {noreply, State}.
-
-terminate(_Reason, State = #pstate{log_handle = LogHandle}) ->
-    flush(State),
-    disk_log:close(LogHandle),
-    ok.
-
-code_change(_OldVsn, State, _Extra) ->
-    {ok, flush(State)}.
-
-%%--------------------------------------------------------------------
-
-internal_extend(Key, MessageList, State) ->
-    log_work(fun (ML) -> {extend_transaction, Key, ML} end,
-             MessageList, State).
-
-internal_dirty_work(MessageList, State) ->
-    log_work(fun (ML) -> {dirty_work, ML} end,
-             MessageList, State).
-
-internal_commit(From, Key, State = #pstate{snapshot = Snapshot}) ->
-    Unit = {commit_transaction, Key},
-    NewSnapshot = internal_integrate1(Unit, Snapshot),
-    complete(From, Unit, State#pstate{snapshot = NewSnapshot}).
-
-internal_rollback(Key, State = #pstate{snapshot = Snapshot}) ->
-    Unit = {rollback_transaction, Key},
-    NewSnapshot = internal_integrate1(Unit, Snapshot),
-    log(State#pstate{snapshot = NewSnapshot}, Unit).
-
-complete(From, Item, State = #pstate{deadline = ExistingDeadline,
-                                     pending_logs = Logs,
-                                     pending_replies = Waiting}) ->
-    State#pstate{deadline = compute_deadline(
-                              ?COMPLETE_BUNDLE_DELAY, ExistingDeadline),
-                 pending_logs = [Item | Logs],
-                 pending_replies = [From | Waiting]}.
-
-%% This is made to limit disk usage by writing messages only once onto
-%% disk.  We keep a table associating pkeys to messages, and provided
-%% the list of messages to output is left to right, we can guarantee
-%% that pkeys will be a backreference to a message in memory when a
-%% "tied" is met.
-log_work(CreateWorkUnit, MessageList,
-         State = #pstate{
-           snapshot = Snapshot = #psnapshot{messages = Messages}}) ->
-    Unit = CreateWorkUnit(
-             rabbit_misc:map_in_order(
-               fun (M = {publish, Message, MsgProps, QK = {_QName, PKey}}) ->
-                       case ets:lookup(Messages, PKey) of
-                           [_] -> {tied, MsgProps, QK};
-                           []  -> ets:insert(Messages, {PKey, Message}),
-                                  M
-                       end;
-                  (M) -> M
-               end,
-               MessageList)),
-    NewSnapshot = internal_integrate1(Unit, Snapshot),
-    log(State#pstate{snapshot = NewSnapshot}, Unit).
-
-log(State = #pstate{deadline = ExistingDeadline, pending_logs = Logs},
-    Message) ->
-    State#pstate{deadline = compute_deadline(?LOG_BUNDLE_DELAY,
-                                             ExistingDeadline),
-                 pending_logs = [Message | Logs]}.
-
-base_filename() ->
-    rabbit_mnesia:dir() ++ "/rabbit_persister.LOG".
-
-take_snapshot(LogHandle, OldFileName, Snapshot) ->
-    ok = disk_log:sync(LogHandle),
-    %% current_snapshot is the Head (ie. first thing logged)
-    ok = disk_log:reopen(LogHandle, OldFileName, current_snapshot(Snapshot)).
-
-take_snapshot(LogHandle, Snapshot) ->
-    OldFileName = lists:flatten(base_filename() ++ ".previous"),
-    file:delete(OldFileName),
-    rabbit_log:info("Rolling persister log to ~p~n", [OldFileName]),
-    ok = take_snapshot(LogHandle, OldFileName, Snapshot).
-
-take_snapshot_and_save_old(LogHandle, Snapshot) ->
-    {MegaSecs, Secs, MicroSecs} = erlang:now(),
-    Timestamp = MegaSecs * 1000000 + Secs * 1000 + MicroSecs,
-    OldFileName = lists:flatten(io_lib:format("~s.saved.~p",
-                                              [base_filename(), Timestamp])),
-    rabbit_log:info("Saving persister log in ~p~n", [OldFileName]),
-    ok = take_snapshot(LogHandle, OldFileName, Snapshot).
-
-maybe_take_snapshot(Force, State = #pstate{entry_count = EntryCount,
-                                           log_handle = LH,
-                                           snapshot = Snapshot}) ->
-    {ok, MaxWrapEntries} = application:get_env(persister_max_wrap_entries),
-    if
-        Force orelse EntryCount >= MaxWrapEntries ->
-            ok = take_snapshot(LH, Snapshot),
-            State#pstate{entry_count = 0};
-        true ->
-            State
-    end.
-
-later_ms(DeltaMilliSec) ->
-    {MegaSec, Sec, MicroSec} = now(),
-    %% Note: not normalised. Unimportant for this application.
-    {MegaSec, Sec, MicroSec + (DeltaMilliSec * 1000)}.
-
-%% Result = B - A, more or less
-time_diff({B1, B2, B3}, {A1, A2, A3}) ->
-    (B1 - A1) * 1000000 + (B2 - A2) + (B3 - A3) / 1000000.0 .
-
-compute_deadline(TimerDelay, infinity) ->
-    later_ms(TimerDelay);
-compute_deadline(_TimerDelay, ExistingDeadline) ->
-    ExistingDeadline.
-
-compute_timeout(infinity) ->
-    {ok, HibernateAfter} = application:get_env(persister_hibernate_after),
-    HibernateAfter;
-compute_timeout(Deadline) ->
-    DeltaMilliSec = time_diff(Deadline, now()) * 1000.0,
-    if
-        DeltaMilliSec =< 1 ->
-            0;
-        true ->
-            round(DeltaMilliSec)
-    end.
-
-do_noreply(State = #pstate{deadline = Deadline}) ->
-    {noreply, State, compute_timeout(Deadline)}.
-
-do_reply(Reply, State = #pstate{deadline = Deadline}) ->
-    {reply, Reply, State, compute_timeout(Deadline)}.
-
-flush(State) -> flush(false, State).
-
-flush(ForceSnapshot, State = #pstate{pending_logs = PendingLogs,
-                                     pending_replies = Waiting,
-                                     log_handle = LogHandle}) ->
-    State1 = if PendingLogs /= [] ->
-                     disk_log:alog(LogHandle, lists:reverse(PendingLogs)),
-                     State#pstate{entry_count = State#pstate.entry_count + 1};
-                true ->
-                     State
-             end,
-    State2 = maybe_take_snapshot(ForceSnapshot, State1),
-    if Waiting /= [] ->
-            ok = disk_log:sync(LogHandle),
-            lists:foreach(fun (From) -> gen_server:reply(From, ok) end,
-                          Waiting);
-       true ->
-            ok
-    end,
-    State2#pstate{deadline = infinity,
-                  pending_logs = [],
-                  pending_replies = []}.
-
-current_snapshot(_Snapshot = #psnapshot{transactions = Ts,
-                                        messages     = Messages,
-                                        queues       = Queues,
-                                        next_seq_id  = NextSeqId}) ->
-    %% Avoid infinite growth of the table by removing messages not
-    %% bound to a queue anymore
-    PKeys = ets:foldl(fun ({{_QName, PKey}, _Delivered, 
-                            _MsgProps, _SeqId}, S) ->
-                              sets:add_element(PKey, S)
-                      end, sets:new(), Queues),
-    prune_table(Messages, fun (Key) -> sets:is_element(Key, PKeys) end),
-    InnerSnapshot = {{txns, Ts},
-                     {messages, ets:tab2list(Messages)},
-                     {queues, ets:tab2list(Queues)},
-                     {next_seq_id, NextSeqId}},
-    ?LOGDEBUG("Inner snapshot: ~p~n", [InnerSnapshot]),
-    {persist_snapshot, {vsn, ?PERSISTER_LOG_FORMAT_VERSION},
-     term_to_binary(InnerSnapshot)}.
-
-prune_table(Tab, Pred) ->
-    true = ets:safe_fixtable(Tab, true),
-    ok = prune_table(Tab, Pred, ets:first(Tab)),
-    true = ets:safe_fixtable(Tab, false).
-
-prune_table(_Tab, _Pred, '$end_of_table') -> ok;
-prune_table(Tab, Pred, Key) ->
-    case Pred(Key) of
-        true  -> ok;
-        false -> ets:delete(Tab, Key)
-    end,
-    prune_table(Tab, Pred, ets:next(Tab, Key)).
-
-internal_load_snapshot(LogHandle,
-                       DurableQueues,
-                       Snapshot = #psnapshot{messages = Messages,
-                                             queues = Queues}) ->
-    {K, [Loaded_Snapshot | Items]} = disk_log:chunk(LogHandle, start),
-    case check_version(Loaded_Snapshot) of
-        {ok, StateBin} ->
-            {{txns, Ts}, {messages, Ms}, {queues, Qs},
-             {next_seq_id, NextSeqId}} = binary_to_term(StateBin),
-            true = ets:insert(Messages, Ms),
-            true = ets:insert(Queues, Qs),
-            Snapshot1 = replay(Items, LogHandle, K,
-                               Snapshot#psnapshot{
-                                 transactions = Ts,
-                                 next_seq_id = NextSeqId}),
-            %% Remove all entries for queues that no longer exist.
-            %% Note that the 'messages' table is pruned when the next
-            %% snapshot is taken.
-            DurableQueuesSet = sets:from_list(DurableQueues),
-            prune_table(Snapshot1#psnapshot.queues,
-                        fun ({QName, _PKey}) ->
-                                sets:is_element(QName, DurableQueuesSet)
-                        end),
-            %% uncompleted transactions are discarded - this is TRTTD
-            %% since we only get into this code on node restart, so
-            %% any uncompleted transactions will have been aborted.
-            {ok, Snapshot1#psnapshot{transactions = dict:new()}};
-        {error, Reason} -> {{error, Reason}, Snapshot}
-    end.
-
-check_version({persist_snapshot, {vsn, ?PERSISTER_LOG_FORMAT_VERSION},
-               StateBin}) ->
-    {ok, StateBin};
-check_version({persist_snapshot, {vsn, Vsn}, _StateBin}) ->
-    {error, {unsupported_persister_log_format, Vsn}};
-check_version(_Other) ->
-    {error, unrecognised_persister_log_format}.
-
-replay([], LogHandle, K, Snapshot) ->
-    case disk_log:chunk(LogHandle, K) of
-        {K1, Items} ->
-            replay(Items, LogHandle, K1, Snapshot);
-        {K1, Items, Badbytes} ->
-            rabbit_log:warning("~p bad bytes recovering persister log~n",
-                               [Badbytes]),
-            replay(Items, LogHandle, K1, Snapshot);
-        eof -> Snapshot
-    end;
-replay([Item | Items], LogHandle, K, Snapshot) ->
-    NewSnapshot = internal_integrate_messages(Item, Snapshot),
-    replay(Items, LogHandle, K, NewSnapshot).
-
-internal_integrate_messages(Items, Snapshot) ->
-    lists:foldl(fun (Item, Snap) -> internal_integrate1(Item, Snap) end,
-                Snapshot, Items).
-
-internal_integrate1({extend_transaction, Key, MessageList},
-                    Snapshot = #psnapshot {transactions = Transactions}) ->
-    Snapshot#psnapshot{transactions = rabbit_misc:dict_cons(Key, MessageList,
-                                                            Transactions)};
-internal_integrate1({rollback_transaction, Key},
-                    Snapshot = #psnapshot{transactions = Transactions}) ->
-    Snapshot#psnapshot{transactions = dict:erase(Key, Transactions)};
-internal_integrate1({commit_transaction, Key},
-                    Snapshot = #psnapshot{transactions = Transactions,
-                                          messages     = Messages,
-                                          queues       = Queues,
-                                          next_seq_id  = SeqId}) ->
-    case dict:find(Key, Transactions) of
-        {ok, MessageLists} ->
-            ?LOGDEBUG("persist committing txn ~p~n", [Key]),
-            NextSeqId =
-                lists:foldr(
-                  fun (ML, SeqIdN) ->
-                          perform_work(ML, Messages, Queues, SeqIdN) end,
-                  SeqId, MessageLists),
-            Snapshot#psnapshot{transactions = dict:erase(Key, Transactions),
-                               next_seq_id = NextSeqId};
-        error ->
-            Snapshot
-    end;
-internal_integrate1({dirty_work, MessageList},
-                    Snapshot = #psnapshot{messages    = Messages,
-                                          queues      = Queues,
-                                          next_seq_id = SeqId}) ->
-    Snapshot#psnapshot{next_seq_id = perform_work(MessageList, Messages,
-                                                  Queues, SeqId)}.
-
-perform_work(MessageList, Messages, Queues, SeqId) ->
-    lists:foldl(fun (Item, NextSeqId) ->
-                        perform_work_item(Item, Messages, Queues, NextSeqId)
-                end, SeqId, MessageList).
-
-perform_work_item({publish, Message, MsgProps, QK = {_QName, PKey}},
-                  Messages, Queues, NextSeqId) ->
-    true = ets:insert(Messages, {PKey, Message}),
-    true = ets:insert(Queues, {QK, false, MsgProps, NextSeqId}),
-    NextSeqId + 1;
-
-perform_work_item({tied, MsgProps, QK}, _Messages, Queues, NextSeqId) ->
-    true = ets:insert(Queues, {QK, false, MsgProps, NextSeqId}),
-    NextSeqId + 1;
-
-perform_work_item({deliver, QK}, _Messages, Queues, NextSeqId) ->
-    true = ets:update_element(Queues, QK, {2, true}),
-    NextSeqId;
-
-perform_work_item({ack, QK}, _Messages, Queues, NextSeqId) ->
-    true = ets:delete(Queues, QK),
-    NextSeqId.
diff --git a/src/rabbit_plugin_activator.erl b/src/rabbit_prelaunch.erl
index ef81ddf2..867ecb12 100644
--- a/src/rabbit_plugin_activator.erl
+++ b/src/rabbit_prelaunch.erl
@@ -29,11 +29,12 @@
 %%   Contributor(s): ______________________________________.
 %%
 
--module(rabbit_plugin_activator).
+-module(rabbit_prelaunch).
 
 -export([start/0, stop/0]).
 
 -define(BaseApps, [rabbit]).
+-define(ERROR_CODE, 1).
 
 %%----------------------------------------------------------------------------
 %% Specs
@@ -50,13 +51,9 @@
 
 start() ->
     io:format("Activating RabbitMQ plugins ...~n"),
-    %% Ensure Rabbit is loaded so we can access it's environment
-    application:load(rabbit),
 
     %% Determine our various directories
-    {ok, PluginDir}         = application:get_env(rabbit, plugins_dir),
-    {ok, UnpackedPluginDir} = application:get_env(rabbit, plugins_expand_dir),
-
+    [PluginDir, UnpackedPluginDir, Node] = init:get_plain_arguments(),
     RootName = UnpackedPluginDir ++ "/rabbit",
 
     %% Unpack any .ez plugins
@@ -134,7 +131,10 @@ start() ->
     [io:format("* ~s-~s~n", [App, proplists:get_value(App, AppVersions)])
      || App <- PluginApps],
     io:nl(),
-    halt(),
+
+    ok = duplicate_node_check(Node),
+
+    terminate(0),
     ok.
 
 stop() ->
@@ -255,6 +255,40 @@ process_entry(Entry = {apply,{application,start_boot,[rabbit,permanent]}}) ->
 process_entry(Entry) ->
     [Entry].
 
+%% Check whether a node with the same name is already running
+duplicate_node_check([]) ->
+    %% Ignore running node while installing windows service
+    ok;
+duplicate_node_check(Node) ->
+    {NodeName, NodeHost} = rabbit_misc:nodeparts(Node),
+    case net_adm:names(NodeHost) of
+        {ok, NamePorts}  ->
+            case proplists:is_defined(NodeName, NamePorts) of
+                     true -> io:format("node with name ~p "
+                                       "already running on ~p~n",
+                                       [NodeName, NodeHost]),
+                             [io:format(Fmt ++ "~n", Args) ||
+                              {Fmt, Args} <- rabbit_control:diagnostics(Node)],
+                             terminate(?ERROR_CODE);
+                     false -> ok
+            end;
+        {error, address}    -> ok;
+        {error, EpmdReason} -> terminate("unexpected epmd error: ~p~n",
+                                         [EpmdReason])
+    end.
+
 terminate(Fmt, Args) ->
     io:format("ERROR: " ++ Fmt ++ "~n", Args),
-    halt(1).
+    terminate(?ERROR_CODE).
+
+terminate(Status) ->
+    case os:type() of
+        {unix, _} ->
+            halt(Status);
+        {win32, _} ->
+            init:stop(Status),
+            receive
+                after infinity -> ok
+            end
+    end.
+
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index bde9b3d3..76c0a4ef 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -31,11 +31,13 @@
 
 -module(rabbit_queue_index).
 
--export([init/1, shutdown_terms/1, recover/4,
+-export([init/2, shutdown_terms/1, recover/5,
          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]).
 
+-export([add_queue_ttl/0]).
+
 -define(CLEAN_FILENAME, "clean.dot").
 
 %%----------------------------------------------------------------------------
@@ -172,7 +174,7 @@
 %%----------------------------------------------------------------------------
 
 -record(qistate, { dir, segments, journal_handle, dirty_count,
-                   max_journal_entries }).
+                   max_journal_entries, on_sync, unsynced_guids }).
 
 -record(segment, { num, path, journal_entries, unacked }).
 
@@ -180,6 +182,8 @@
 
 %%----------------------------------------------------------------------------
 
+-rabbit_upgrade({add_queue_ttl, []}).
+
 -ifdef(use_specs).
 
 -type(hdl() :: ('undefined' | any())).
@@ -191,21 +195,24 @@
                               })).
 -type(seq_id() :: integer()).
 -type(seg_dict() :: {dict(), [segment()]}).
+-type(on_sync_fun() :: fun ((gb_set()) -> ok)).
 -type(qistate() :: #qistate { dir                 :: file:filename(),
                               segments            :: 'undefined' | seg_dict(),
                               journal_handle      :: hdl(),
                               dirty_count         :: integer(),
-                              max_journal_entries :: non_neg_integer()
+                              max_journal_entries :: non_neg_integer(),
+                              on_sync             :: on_sync_fun(),
+                              unsynced_guids      :: [rabbit_guid:guid()]
                              }).
 -type(startup_fun_state() ::
-        {(fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A})),
+        {fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A}),
          A}).
 -type(shutdown_terms() :: [any()]).
 
--spec(init/1 :: (rabbit_amqqueue:name()) -> qistate()).
+-spec(init/2 :: (rabbit_amqqueue:name(), on_sync_fun()) -> qistate()).
 -spec(shutdown_terms/1 :: (rabbit_amqqueue:name()) -> shutdown_terms()).
--spec(recover/4 :: (rabbit_amqqueue:name(), shutdown_terms(), boolean(),
-                    fun ((rabbit_guid:guid()) -> boolean())) ->
+-spec(recover/5 :: (rabbit_amqqueue:name(), shutdown_terms(), boolean(),
+                    fun ((rabbit_guid:guid()) -> boolean()), on_sync_fun()) ->
              {'undefined' | non_neg_integer(), qistate()}).
 -spec(terminate/2 :: ([any()], qistate()) -> qistate()).
 -spec(delete_and_terminate/1 :: (qistate()) -> qistate()).
@@ -223,8 +230,10 @@
 -spec(next_segment_boundary/1 :: (seq_id()) -> seq_id()).
 -spec(bounds/1 :: (qistate()) ->
              {non_neg_integer(), non_neg_integer(), qistate()}).
--spec(recover/1 ::
-        ([rabbit_amqqueue:name()]) -> {[[any()]], startup_fun_state()}).
+-spec(recover/1 :: ([rabbit_amqqueue:name()]) ->
+                        {[[any()]], startup_fun_state()}).
+
+-spec(add_queue_ttl/0 :: () -> 'ok').
 
 -endif.
 
@@ -233,10 +242,10 @@
 %% public API
 %%----------------------------------------------------------------------------
 
-init(Name) ->
+init(Name, OnSyncFun) ->
     State = #qistate { dir = Dir } = blank_state(Name),
     false = filelib:is_file(Dir), %% is_file == is file or dir
-    State.
+    State #qistate { on_sync = OnSyncFun }.
 
 shutdown_terms(Name) ->
     #qistate { dir = Dir } = blank_state(Name),
@@ -245,13 +254,14 @@ shutdown_terms(Name) ->
         {ok, Terms1} -> Terms1
     end.
 
-recover(Name, Terms, MsgStoreRecovered, ContainsCheckFun) ->
+recover(Name, Terms, MsgStoreRecovered, ContainsCheckFun, OnSyncFun) ->
     State = #qistate { dir = Dir } = blank_state(Name),
+    State1 = State #qistate { on_sync = OnSyncFun },
     CleanShutdown = detect_clean_shutdown(Dir),
     case CleanShutdown andalso MsgStoreRecovered of
         true  -> RecoveredCounts = proplists:get_value(segments, Terms, []),
-                 init_clean(RecoveredCounts, State);
-        false -> init_dirty(CleanShutdown, ContainsCheckFun, State)
+                 init_clean(RecoveredCounts, State1);
+        false -> init_dirty(CleanShutdown, ContainsCheckFun, State1)
     end.
 
 terminate(Terms, State) ->
@@ -264,9 +274,13 @@ delete_and_terminate(State) ->
     ok = rabbit_misc:recursive_delete([Dir]),
     State1.
 
-publish(Guid, SeqId, MsgProps, IsPersistent, State) when is_binary(Guid) ->
+publish(Guid, SeqId, MsgProps, IsPersistent,
+        State = #qistate { unsynced_guids = UnsyncedGuids })
+  when is_binary(Guid) ->
     ?GUID_BYTES = size(Guid),
-    {JournalHdl, State1} = get_journal_handle(State),
+    {JournalHdl, State1} = get_journal_handle(
+                             State #qistate {
+                               unsynced_guids = [Guid | UnsyncedGuids] }),
     ok = file_handle_cache:append(
            JournalHdl, [<<(case IsPersistent of
                                true  -> ?PUB_PERSIST_JPREFIX;
@@ -297,7 +311,7 @@ sync(_SeqIds, State = #qistate { journal_handle = JournalHdl }) ->
     %% seqids not being in the journal, provided the transaction isn't
     %% emptied (handled above anyway).
     ok = file_handle_cache:sync(JournalHdl),
-    State.
+    notify_sync(State).
 
 flush(State = #qistate { dirty_count = 0 }) -> State;
 flush(State)                                -> flush_journal(State).
@@ -345,35 +359,36 @@ recover(DurableQueues) ->
     DurableDict = dict:from_list([ {queue_name_to_dir_name(Queue), Queue} ||
                                      Queue <- DurableQueues ]),
     QueuesDir = queues_dir(),
-    Directories = case file:list_dir(QueuesDir) of
-                      {ok, Entries}   -> [ Entry || Entry <- Entries,
-                                                    filelib:is_dir(
-                                                      filename:join(
-                                                        QueuesDir, Entry)) ];
-                      {error, enoent} -> []
-                  end,
+    QueueDirNames = all_queue_directory_names(QueuesDir),
     DurableDirectories = sets:from_list(dict:fetch_keys(DurableDict)),
     {DurableQueueNames, DurableTerms} =
         lists:foldl(
-          fun (QueueDir, {DurableAcc, TermsAcc}) ->
-                  case sets:is_element(QueueDir, DurableDirectories) of
+          fun (QueueDirName, {DurableAcc, TermsAcc}) ->
+                  QueueDirPath = filename:join(QueuesDir, QueueDirName),
+                  case sets:is_element(QueueDirName, DurableDirectories) of
                       true ->
                           TermsAcc1 =
-                              case read_shutdown_terms(
-                                     filename:join(QueuesDir, QueueDir)) of
+                              case read_shutdown_terms(QueueDirPath) of
                                   {error, _}  -> TermsAcc;
                                   {ok, Terms} -> [Terms | TermsAcc]
                               end,
-                          {[dict:fetch(QueueDir, DurableDict) | DurableAcc],
+                          {[dict:fetch(QueueDirName, DurableDict) | DurableAcc],
                            TermsAcc1};
                       false ->
-                          Dir = filename:join(queues_dir(), QueueDir),
-                          ok = rabbit_misc:recursive_delete([Dir]),
+                          ok = rabbit_misc:recursive_delete([QueueDirPath]),
                           {DurableAcc, TermsAcc}
                   end
-          end, {[], []}, Directories),
+          end, {[], []}, QueueDirNames),
     {DurableTerms, {fun queue_index_walker/1, {start, DurableQueueNames}}}.
 
+all_queue_directory_names(Dir) ->
+    case file:list_dir(Dir) of
+        {ok, Entries}   -> [ Entry || Entry <- Entries,
+                                      filelib:is_dir(
+                                        filename:join(Dir, Entry)) ];
+        {error, enoent} -> []
+    end.
+
 %%----------------------------------------------------------------------------
 %% startup and shutdown
 %%----------------------------------------------------------------------------
@@ -386,7 +401,9 @@ blank_state(QueueName) ->
                segments            = segments_new(),
                journal_handle      = undefined,
                dirty_count         = 0,
-               max_journal_entries = MaxJournal }.
+               max_journal_entries = MaxJournal,
+               on_sync             = fun (_) -> ok end,
+               unsynced_guids      = [] }.
 
 clean_file_name(Dir) -> filename:join(Dir, ?CLEAN_FILENAME).
 
@@ -618,7 +635,7 @@ flush_journal(State = #qistate { segments = Segments }) ->
     {JournalHdl, State1} =
         get_journal_handle(State #qistate { segments = Segments1 }),
     ok = file_handle_cache:clear(JournalHdl),
-    State1 #qistate { dirty_count = 0 }.
+    notify_sync(State1 #qistate { dirty_count = 0 }).
 
 append_journal_to_segment(#segment { journal_entries = JEntries,
                                      path = Path } = Segment) ->
@@ -706,6 +723,10 @@ deliver_or_ack(Kind, SeqIds, State) ->
                                             add_to_journal(SeqId, Kind, StateN)
                                     end, State1, SeqIds)).
 
+notify_sync(State = #qistate { unsynced_guids = UG, on_sync = OnSyncFun }) ->
+    OnSyncFun(gb_sets:from_list(UG)),
+    State #qistate { unsynced_guids = [] }.
+
 %%----------------------------------------------------------------------------
 %% segment manipulation
 %%----------------------------------------------------------------------------
@@ -972,3 +993,86 @@ journal_minus_segment1({no_pub, del, ack},         {?PUB, del, no_ack}) ->
     {{no_pub, no_del, ack}, 0};
 journal_minus_segment1({no_pub, del, ack},         {?PUB, del, ack}) ->
     {undefined, -1}.
+
+%%----------------------------------------------------------------------------
+%% upgrade
+%%----------------------------------------------------------------------------
+
+add_queue_ttl() ->
+    foreach_queue_index({fun add_queue_ttl_journal/1,
+                         fun add_queue_ttl_segment/1}).
+
+add_queue_ttl_journal(<<?DEL_JPREFIX:?JPREFIX_BITS, SeqId:?SEQ_BITS,
+                        Rest/binary>>) ->
+    {<<?DEL_JPREFIX:?JPREFIX_BITS, SeqId:?SEQ_BITS>>, Rest};
+add_queue_ttl_journal(<<?ACK_JPREFIX:?JPREFIX_BITS, SeqId:?SEQ_BITS,
+                        Rest/binary>>) ->
+    {<<?ACK_JPREFIX:?JPREFIX_BITS, SeqId:?SEQ_BITS>>, Rest};
+add_queue_ttl_journal(<<Prefix:?JPREFIX_BITS, SeqId:?SEQ_BITS,
+                        Guid:?GUID_BYTES/binary, Rest/binary>>) ->
+    {[<<Prefix:?JPREFIX_BITS, SeqId:?SEQ_BITS>>, Guid,
+      expiry_to_binary(undefined)], Rest};
+add_queue_ttl_journal(_) ->
+    stop.
+
+add_queue_ttl_segment(<<?PUBLISH_PREFIX:?PUBLISH_PREFIX_BITS, IsPersistentNum:1,
+                        RelSeq:?REL_SEQ_BITS, Guid:?GUID_BYTES/binary,
+                        Rest/binary>>) ->
+    {[<<?PUBLISH_PREFIX:?PUBLISH_PREFIX_BITS, IsPersistentNum:1,
+        RelSeq:?REL_SEQ_BITS>>, Guid, expiry_to_binary(undefined)], Rest};
+add_queue_ttl_segment(<<?REL_SEQ_ONLY_PREFIX:?REL_SEQ_ONLY_PREFIX_BITS,
+                        RelSeq:?REL_SEQ_BITS, Rest>>) ->
+    {<<?REL_SEQ_ONLY_PREFIX:?REL_SEQ_ONLY_PREFIX_BITS, RelSeq:?REL_SEQ_BITS>>,
+     Rest};
+add_queue_ttl_segment(_) ->
+    stop.
+
+%%----------------------------------------------------------------------------
+
+foreach_queue_index(Funs) ->
+    QueuesDir = queues_dir(),
+    QueueDirNames = all_queue_directory_names(QueuesDir),
+    {ok, Gatherer} = gatherer:start_link(),
+    [begin
+         ok = gatherer:fork(Gatherer),
+         ok = worker_pool:submit_async(
+                fun () ->
+                        transform_queue(filename:join(QueuesDir, QueueDirName),
+                                        Gatherer, Funs)
+                end)
+     end || QueueDirName <- QueueDirNames],
+    empty = gatherer:out(Gatherer),
+    ok = gatherer:stop(Gatherer),
+    ok = rabbit_misc:unlink_and_capture_exit(Gatherer).
+
+transform_queue(Dir, Gatherer, {JournalFun, SegmentFun}) ->
+    ok = transform_file(filename:join(Dir, ?JOURNAL_FILENAME), JournalFun),
+    [ok = transform_file(filename:join(Dir, Seg), SegmentFun)
+     || Seg <- filelib:wildcard("*" ++ ?SEGMENT_EXTENSION, Dir)],
+    ok = gatherer:finish(Gatherer).
+
+transform_file(Path, Fun) ->
+    PathTmp = Path ++ ".upgrade",
+    case filelib:file_size(Path) of
+        0    -> ok;
+        Size -> {ok, PathTmpHdl} =
+                    file_handle_cache:open(PathTmp, ?WRITE_MODE,
+                                           [{write_buffer, infinity}]),
+
+                {ok, PathHdl} = file_handle_cache:open(
+                                  Path, [{read_ahead, Size} | ?READ_MODE], []),
+                {ok, Content} = file_handle_cache:read(PathHdl, Size),
+                ok = file_handle_cache:close(PathHdl),
+
+                ok = drive_transform_fun(Fun, PathTmpHdl, Content),
+
+                ok = file_handle_cache:close(PathTmpHdl),
+                ok = file:rename(PathTmp, Path)
+    end.
+
+drive_transform_fun(Fun, Hdl, Contents) ->
+    case Fun(Contents) of
+        stop                -> ok;
+        {Output, Contents1} -> ok = file_handle_cache:append(Hdl, Output),
+                               drive_transform_fun(Fun, Hdl, Contents1)
+    end.
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 12730ccf..4dd150a2 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -45,10 +45,6 @@
 
 -export([emit_stats/1]).
 
--import(gen_tcp).
--import(inet).
--import(prim_inet).
-
 -define(HANDSHAKE_TIMEOUT, 10).
 -define(NORMAL_TIMEOUT, 3).
 -define(CLOSING_TIMEOUT, 1).
@@ -322,13 +318,10 @@ start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
     done.
 
 mainloop(Deb, State = #v1{parent = Parent, sock= Sock, recv_ref = Ref}) ->
-    %%?LOGDEBUG("Reader mainloop: ~p bytes available, need ~p~n", [HaveBytes, WaitUntilNBytes]),
     receive
         {inet_async, Sock, Ref, {ok, Data}} ->
-            {State1, Callback1, Length1} =
-                handle_input(State#v1.callback, Data,
-                             State#v1{recv_ref = none}),
-            mainloop(Deb, switch_callback(State1, Callback1, Length1));
+            mainloop(Deb, handle_input(State#v1.callback, Data,
+                                       State#v1{recv_ref = none}));
         {inet_async, Sock, Ref, {error, closed}} ->
             if State#v1.connection_state =:= closed ->
                     State;
@@ -568,7 +561,6 @@ handle_frame(Type, Channel, Payload,
         error         -> throw({unknown_frame, Channel, Type, Payload});
         heartbeat     -> throw({unexpected_heartbeat_frame, Channel});
         AnalyzedFrame ->
-            %%?LOGDEBUG("Ch ~p Frame ~p~n", [Channel, AnalyzedFrame]),
             case get({channel, Channel}) of
                 {ch_fr_pid, ChFrPid} ->
                     ok = rabbit_framing_channel:process(ChFrPid, AnalyzedFrame),
@@ -598,8 +590,9 @@ handle_frame(Type, Channel, Payload,
                             %% We're already closing this channel, so
                             %% there's no cleanup to do (notify
                             %% queues, etc.)
-                            ok = rabbit_writer:send_command(State#v1.sock,
-                                                            #'channel.close_ok'{});
+                            ok = rabbit_writer:internal_send_command(
+                                   State#v1.sock, Channel,
+                                   #'channel.close_ok'{}, Protocol);
                         _ -> ok
                     end,
                     State;
@@ -632,18 +625,18 @@ analyze_frame(_Type, _Body, _Protocol) ->
     error.
 
 handle_input(frame_header, <<Type:8,Channel:16,PayloadSize:32>>, State) ->
-    %%?LOGDEBUG("Got frame header: ~p/~p/~p~n", [Type, Channel, PayloadSize]),
-    {ensure_stats_timer(State), {frame_payload, Type, Channel, PayloadSize},
-     PayloadSize + 1};
+    ensure_stats_timer(
+      switch_callback(State, {frame_payload, Type, Channel, PayloadSize},
+                      PayloadSize + 1));
 
-handle_input({frame_payload, Type, Channel, PayloadSize}, PayloadAndMarker, State) ->
+handle_input({frame_payload, Type, Channel, PayloadSize},
+             PayloadAndMarker, State) ->
     case PayloadAndMarker of
         <<Payload:PayloadSize/binary, ?FRAME_END>> ->
-            %%?LOGDEBUG("Frame completed: ~p/~p/~p~n", [Type, Channel, Payload]),
-            NewState = handle_frame(Type, Channel, Payload, State),
-            {NewState, frame_header, 7};
+            handle_frame(Type, Channel, Payload,
+                         switch_callback(State, frame_header, 7));
         _ ->
-            throw({bad_payload, PayloadAndMarker})
+            throw({bad_payload, Type, Channel, PayloadSize, PayloadAndMarker})
     end;
 
 %% The two rules pertaining to version negotiation:
@@ -694,11 +687,11 @@ start_connection({ProtocolMajor, ProtocolMinor, _ProtocolRevision},
                                  mechanisms = <<"PLAIN AMQPLAIN">>,
                                  locales = <<"en_US">> },
     ok = send_on_channel0(Sock, Start, Protocol),
-    {State#v1{connection = Connection#connection{
-                             timeout_sec = ?NORMAL_TIMEOUT,
-                             protocol = Protocol},
-              connection_state = starting},
-     frame_header, 7}.
+    switch_callback(State#v1{connection = Connection#connection{
+                                            timeout_sec = ?NORMAL_TIMEOUT,
+                                            protocol = Protocol},
+                             connection_state = starting},
+                    frame_header, 7).
 
 refuse_connection(Sock, Exception) ->
     ok = inet_op(fun () -> rabbit_net:send(Sock, <<"AMQP",0,0,9,1>>) end),
diff --git a/src/rabbit_router.erl b/src/rabbit_router.erl
index 00df1ce1..d49c072c 100644
--- a/src/rabbit_router.erl
+++ b/src/rabbit_router.erl
@@ -72,7 +72,8 @@ deliver(QNames, Delivery = #delivery{mandatory = false,
       QPids, fun (Pid) -> rabbit_amqqueue:deliver(Pid, Delivery) end),
     {routed, QPids};
 
-deliver(QNames, Delivery) ->
+deliver(QNames, Delivery = #delivery{mandatory = Mandatory,
+                                    immediate = Immediate}) ->
     QPids = lookup_qpids(QNames),
     {Success, _} =
         delegate:invoke(QPids,
@@ -80,9 +81,9 @@ deliver(QNames, Delivery) ->
                                 rabbit_amqqueue:deliver(Pid, Delivery)
                         end),
     {Routed, Handled} =
-        lists:foldl(fun fold_deliveries/2, {false, []}, Success),
-    check_delivery(Delivery#delivery.mandatory, Delivery#delivery.immediate,
-                   {Routed, Handled}).
+         lists:foldl(fun fold_deliveries/2, {false, []}, Success),
+    check_delivery(Mandatory, Immediate, {Routed, Handled}).
+
 
 %% TODO: Maybe this should be handled by a cursor instead.
 %% TODO: This causes a full scan for each entry with the same source
diff --git a/src/rabbit_ssl.erl b/src/rabbit_ssl.erl
index 5b905682..1d8ce23b 100644
--- a/src/rabbit_ssl.erl
+++ b/src/rabbit_ssl.erl
@@ -95,14 +95,11 @@ cert_info(F, Cert) ->
 
 %% Format and rdnSequence as a RFC4514 subject string.
 format_rdn_sequence({rdnSequence, Seq}) ->
-    lists:flatten(
-      rabbit_misc:intersperse(
-        ",", lists:reverse([format_complex_rdn(RDN) || RDN <- Seq]))).
+    string:join(lists:reverse([format_complex_rdn(RDN) || RDN <- Seq]), ",").
 
 %% Format an RDN set.
 format_complex_rdn(RDNs) ->
-    lists:flatten(
-      rabbit_misc:intersperse("+", [format_rdn(RDN) || RDN <- RDNs])).
+    string:join([format_rdn(RDN) || RDN <- RDNs], "+").
 
 %% Format an RDN.  If the type name is unknown, use the dotted decimal
 %% representation.  See RFC4514, section 2.3.
@@ -129,7 +126,7 @@ format_rdn(#'AttributeTypeAndValue'{type = T, value = V}) ->
             io_lib:format(Fmt ++ "=~s", [FV]);
         none when is_tuple(T) ->
             TypeL = [io_lib:format("~w", [X]) || X <- tuple_to_list(T)],
-            io_lib:format("~s:~s", [rabbit_misc:intersperse(".", TypeL), FV]);
+            io_lib:format("~s:~s", [string:join(TypeL, "."), FV]);
         none ->
             io_lib:format("~p:~s", [T, FV])
     end.
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 71b23e01..adf968cb 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -35,8 +35,6 @@
 
 -export([all_tests/0, test_parsing/0]).
 
--import(lists).
-
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
 -include_lib("kernel/include/file.hrl").
@@ -1470,12 +1468,12 @@ msg_store_remove(MsgStore, Ref, Guids) ->
 
 with_msg_store_client(MsgStore, Ref, Fun) ->
     rabbit_msg_store:client_terminate(
-      Fun(rabbit_msg_store:client_init(MsgStore, Ref))).
+      Fun(rabbit_msg_store:client_init(MsgStore, Ref, undefined))).
 
 foreach_with_msg_store_client(MsgStore, Ref, Fun, L) ->
     rabbit_msg_store:client_terminate(
       lists:foldl(fun (Guid, MSCState) -> Fun(Guid, MSCState) end,
-                  rabbit_msg_store:client_init(MsgStore, Ref), L)).
+                  rabbit_msg_store:client_init(MsgStore, Ref, undefined), L)).
 
 test_msg_store() ->
     restart_msg_store_empty(),
@@ -1483,7 +1481,8 @@ test_msg_store() ->
     Guids = [guid_bin(M) || M <- lists:seq(1,100)],
     {Guids1stHalf, Guids2ndHalf} = lists:split(50, Guids),
     Ref = rabbit_guid:guid(),
-    MSCState = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
+    MSCState = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref,
+                                            undefined),
     %% 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
@@ -1549,7 +1548,8 @@ test_msg_store() ->
                     ([Guid|GuidsTail]) ->
                         {Guid, 0, GuidsTail}
                 end, Guids2ndHalf}),
-    MSCState5 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
+    MSCState5 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref,
+                                             undefined),
     %% check we have the right msgs left
     lists:foldl(
       fun (Guid, Bool) ->
@@ -1558,7 +1558,8 @@ test_msg_store() ->
     ok = rabbit_msg_store:client_terminate(MSCState5),
     %% restart empty
     restart_msg_store_empty(),
-    MSCState6 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
+    MSCState6 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref,
+                                             undefined),
     %% check we don't contain any of the msgs
     false = msg_store_contains(false, Guids, MSCState6),
     %% publish the first half again
@@ -1566,7 +1567,8 @@ test_msg_store() ->
     %% this should force some sort of sync internally otherwise misread
     ok = rabbit_msg_store:client_terminate(
            msg_store_read(Guids1stHalf, MSCState6)),
-    MSCState7 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref),
+    MSCState7 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref,
+                                             undefined),
     ok = rabbit_msg_store:remove(Guids1stHalf, MSCState7),
     ok = rabbit_msg_store:client_terminate(MSCState7),
     %% restart empty
@@ -1625,12 +1627,13 @@ init_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),
+                                                    PRef, undefined),
     Res = rabbit_queue_index:recover(
             TestQueue, Terms, false,
             fun (Guid) ->
                     rabbit_msg_store:contains(Guid, PersistentClient)
-            end),
+            end,
+            fun nop/1),
     ok = rabbit_msg_store:client_delete_and_terminate(PersistentClient),
     Res.
 
@@ -1658,7 +1661,7 @@ queue_index_publish(SeqIds, Persistent, Qi) ->
                    true  -> ?PERSISTENT_MSG_STORE;
                    false -> ?TRANSIENT_MSG_STORE
                end,
-    MSCState = rabbit_msg_store:client_init(MsgStore, Ref),
+    MSCState = rabbit_msg_store:client_init(MsgStore, Ref, undefined),
     {A, B} =
         lists:foldl(
           fun (SeqId, {QiN, SeqIdsGuidsAcc}) ->
@@ -1850,7 +1853,8 @@ assert_props(List, PropVals) ->
 
 with_fresh_variable_queue(Fun) ->
     ok = empty_test_queue(),
-    VQ = rabbit_variable_queue:init(test_queue(), true, false),
+    VQ = rabbit_variable_queue:init(test_queue(), true, false,
+                                    fun nop/1, fun nop/1),
     S0 = rabbit_variable_queue:status(VQ),
     assert_props(S0, [{q1, 0}, {q2, 0},
                       {delta, {delta, undefined, 0, undefined}},
@@ -1865,9 +1869,39 @@ test_variable_queue() ->
               fun test_variable_queue_partial_segments_delta_thing/1,
               fun test_variable_queue_all_the_bits_not_covered_elsewhere1/1,
               fun test_variable_queue_all_the_bits_not_covered_elsewhere2/1,
-              fun test_dropwhile/1]],
+              fun test_dropwhile/1,
+              fun test_variable_queue_ack_limiting/1]],
     passed.
 
+test_variable_queue_ack_limiting(VQ0) ->
+    %% start by sending in a bunch of messages
+    Len = 1024,
+    VQ1 = variable_queue_publish(false, Len, VQ0),
+
+    %% squeeze and relax queue
+    Churn = Len div 32,
+    VQ2 = publish_fetch_and_ack(Churn, Len, VQ1),
+
+    %% update stats for duration
+    {_Duration, VQ3} = rabbit_variable_queue:ram_duration(VQ2),
+
+    %% fetch half the messages
+    {VQ4, _AckTags} = variable_queue_fetch(Len div 2, false, false, Len, VQ3),
+
+    VQ5 = check_variable_queue_status(VQ4, [{len          , Len div 2},
+                                            {ram_ack_count, Len div 2},
+                                            {ram_msg_count, Len div 2}]),
+
+    %% ensure all acks go to disk on 0 duration target
+    VQ6 = check_variable_queue_status(
+            rabbit_variable_queue:set_ram_duration_target(0, VQ5),
+            [{len, Len div 2},
+             {target_ram_count, 0},
+             {ram_msg_count, 0},
+             {ram_ack_count, 0}]),
+
+    VQ6.
+
 test_dropwhile(VQ0) ->
     Count = 10,
 
@@ -1905,7 +1939,6 @@ test_variable_queue_dynamic_duration_change(VQ0) ->
     %% start by sending in a couple of segments worth
     Len = 2*SegmentSize,
     VQ1 = variable_queue_publish(false, Len, VQ0),
-
     %% squeeze and relax queue
     Churn = Len div 32,
     VQ2 = publish_fetch_and_ack(Churn, Len, VQ1),
@@ -1923,7 +1956,7 @@ test_variable_queue_dynamic_duration_change(VQ0) ->
 
     %% drain
     {VQ8, AckTags} = variable_queue_fetch(Len, false, false, Len, VQ7),
-    VQ9 = rabbit_variable_queue:ack(AckTags, VQ8),
+    {_, VQ9} = rabbit_variable_queue:ack(AckTags, VQ8),
     {empty, VQ10} = rabbit_variable_queue:fetch(true, VQ9),
 
     VQ10.
@@ -1933,7 +1966,8 @@ publish_fetch_and_ack(0, _Len, VQ0) ->
 publish_fetch_and_ack(N, Len, VQ0) ->
     VQ1 = variable_queue_publish(false, 1, VQ0),
     {{_Msg, false, AckTag, Len}, VQ2} = rabbit_variable_queue:fetch(true, VQ1),
-    publish_fetch_and_ack(N-1, Len, rabbit_variable_queue:ack([AckTag], VQ2)).
+    {_, VQ3} = rabbit_variable_queue:ack([AckTag], VQ2),
+    publish_fetch_and_ack(N-1, Len, VQ3).
 
 test_variable_queue_partial_segments_delta_thing(VQ0) ->
     SegmentSize = rabbit_queue_index:next_segment_boundary(0),
@@ -1966,7 +2000,7 @@ test_variable_queue_partial_segments_delta_thing(VQ0) ->
              {len, HalfSegment + 1}]),
     {VQ8, AckTags1} = variable_queue_fetch(HalfSegment + 1, true, false,
                                            HalfSegment + 1, VQ7),
-    VQ9 = rabbit_variable_queue:ack(AckTags ++ AckTags1, VQ8),
+    {_, VQ9} = rabbit_variable_queue:ack(AckTags ++ AckTags1, VQ8),
     %% should be empty now
     {empty, VQ10} = rabbit_variable_queue:fetch(true, VQ9),
     VQ10.
@@ -1995,7 +2029,8 @@ test_variable_queue_all_the_bits_not_covered_elsewhere1(VQ0) ->
     {VQ5, _AckTags1} = variable_queue_fetch(Count, false, false,
                                             Count, VQ4),
     _VQ6 = rabbit_variable_queue:terminate(VQ5),
-    VQ7 = rabbit_variable_queue:init(test_queue(), true, true),
+    VQ7 = rabbit_variable_queue:init(test_queue(), true, true,
+                                     fun nop/1, fun nop/1),
     {{_Msg1, true, _AckTag1, Count1}, VQ8} =
         rabbit_variable_queue:fetch(true, VQ7),
     VQ9 = variable_queue_publish(false, 1, VQ8),
@@ -2011,7 +2046,8 @@ test_variable_queue_all_the_bits_not_covered_elsewhere2(VQ0) ->
     VQ4 = rabbit_variable_queue:requeue(AckTags, fun(X) -> X end, VQ3),
     VQ5 = rabbit_variable_queue:idle_timeout(VQ4),
     _VQ6 = rabbit_variable_queue:terminate(VQ5),
-    VQ7 = rabbit_variable_queue:init(test_queue(), true, true),
+    VQ7 = rabbit_variable_queue:init(test_queue(), true, true,
+                                     fun nop/1, fun nop/1),
     {empty, VQ8} = rabbit_variable_queue:fetch(false, VQ7),
     VQ8.
 
@@ -2041,7 +2077,8 @@ test_queue_recover() ->
               {ok, CountMinusOne, {QName, QPid1, _AckTag, true, _Msg}} =
                   rabbit_amqqueue:basic_get(Q1, self(), false),
               exit(QPid1, shutdown),
-              VQ1 = rabbit_variable_queue:init(QName, true, true),
+              VQ1 = rabbit_variable_queue:init(QName, true, true,
+                                               fun nop/1, fun nop/1),
               {{_Msg1, true, _AckTag1, CountMinusOne}, VQ2} =
                   rabbit_variable_queue:fetch(true, VQ1),
               _VQ3 = rabbit_variable_queue:delete_and_terminate(VQ2),
@@ -2101,3 +2138,5 @@ test_configurable_server_properties() ->
 
     application:set_env(rabbit, server_properties, ServerProperties),
     passed.
+
+nop(_) -> ok.
diff --git a/src/rabbit_types.erl b/src/rabbit_types.erl
index b9993823..548014be 100644
--- a/src/rabbit_types.erl
+++ b/src/rabbit_types.erl
@@ -90,7 +90,8 @@
                 sender    :: pid(),
                 message   :: message()}).
 -type(message_properties() ::
-        #message_properties{expiry :: pos_integer() | 'undefined'}).
+        #message_properties{expiry :: pos_integer() | 'undefined',
+                            needs_confirming :: boolean()}).
 
 %% this is really an abstract type, but dialyzer does not support them
 -type(txn() :: rabbit_guid:guid()).
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index 0071a08a..97a07514 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -32,11 +32,13 @@
 
 -ifdef(use_specs).
 
+-type(step() :: atom()).
+-type(version() :: [step()]).
+
 -spec(maybe_upgrade/0 :: () -> 'ok' | 'version_not_available').
--spec(read_version/0 ::
-        () -> {'ok', [any()]} | rabbit_types:error(any())).
+-spec(read_version/0 :: () -> rabbit_types:ok_or_error2(version(), any())).
 -spec(write_version/0 :: () -> 'ok').
--spec(desired_version/0 :: () -> [atom()]).
+-spec(desired_version/0 :: () -> version()).
 
 -endif.
 
@@ -48,26 +50,25 @@
 maybe_upgrade() ->
     case read_version() of
         {ok, CurrentHeads} ->
-            G = load_graph(),
-            case unknown_heads(CurrentHeads, G) of
-                [] ->
-                    case upgrades_to_apply(CurrentHeads, G) of
-                        []       -> ok;
-                        Upgrades -> apply_upgrades(Upgrades)
-                    end;
-                Unknown ->
-                    exit({future_upgrades_found, Unknown})
-            end,
-            true = digraph:delete(G),
-            ok;
+            with_upgrade_graph(
+              fun (G) ->
+                      case unknown_heads(CurrentHeads, G) of
+                          []      -> case upgrades_to_apply(CurrentHeads, G) of
+                                         []       -> ok;
+                                         Upgrades -> apply_upgrades(Upgrades)
+                                     end;
+                          Unknown -> throw({error,
+                                            {future_upgrades_found, Unknown}})
+                      end
+              end);
         {error, enoent} ->
             version_not_available
     end.
 
 read_version() ->
     case rabbit_misc:read_term_file(schema_filename()) of
-        {ok, [Heads]} -> {ok, Heads};
-        {error, E}    -> {error, E}
+        {ok, [Heads]}    -> {ok, Heads};
+        {error, _} = Err -> Err
     end.
 
 write_version() ->
@@ -75,17 +76,26 @@ write_version() ->
     ok.
 
 desired_version() ->
-    G = load_graph(),
-    Version = heads(G),
-    true = digraph:delete(G),
-    Version.
+    with_upgrade_graph(fun (G) -> heads(G) end).
 
 %% -------------------------------------------------------------------
 
-load_graph() ->
-    Upgrades = rabbit_misc:all_module_attributes(rabbit_upgrade),
-    rabbit_misc:build_acyclic_graph(
-      fun vertices/2, fun edges/2, fun graph_build_error/1, Upgrades).
+with_upgrade_graph(Fun) ->
+    case rabbit_misc:build_acyclic_graph(
+           fun vertices/2, fun edges/2,
+           rabbit_misc:all_module_attributes(rabbit_upgrade)) of
+        {ok, G} -> try
+                       Fun(G)
+                   after
+                       true = digraph:delete(G)
+                   end;
+        {error, {vertex, duplicate, StepName}} ->
+            throw({error, {duplicate_upgrade_step, StepName}});
+        {error, {edge, {bad_vertex, StepName}, _From, _To}} ->
+            throw({error, {dependency_on_unknown_upgrade_step, StepName}});
+        {error, {edge, {bad_edge, StepNames}, _From, _To}} ->
+            throw({error, {cycle_in_upgrade_steps, StepNames}})
+    end.
 
 vertices(Module, Steps) ->
     [{StepName, {Module, StepName}} || {StepName, _Reqs} <- Steps].
@@ -93,10 +103,6 @@ vertices(Module, Steps) ->
 edges(_Module, Steps) ->
     [{Require, StepName} || {StepName, Requires} <- Steps, Require <- Requires].
 
-graph_build_error({vertex, duplicate, StepName}) ->
-    exit({duplicate_upgrade, StepName});
-graph_build_error({edge, E, From, To}) ->
-    exit({E, From, To}).
 
 unknown_heads(Heads, G) ->
     [H || H <- Heads, digraph:vertex(G, H) =:= false].
@@ -111,8 +117,8 @@ upgrades_to_apply(Heads, G) ->
                   sets:from_list(digraph_utils:reaching(Heads, G)))),
     %% Form a subgraph from that list and find a topological ordering
     %% so we can invoke them in order.
-    [element(2, digraph:vertex(G, StepName))
-     || StepName <- digraph_utils:topsort(digraph_utils:subgraph(G, Unsorted))].
+    [element(2, digraph:vertex(G, StepName)) ||
+        StepName <- digraph_utils:topsort(digraph_utils:subgraph(G, Unsorted))].
 
 heads(G) ->
     lists:sort([V || V <- digraph:vertices(G), digraph:out_degree(G, V) =:= 0]).
@@ -120,19 +126,35 @@ heads(G) ->
 %% -------------------------------------------------------------------
 
 apply_upgrades(Upgrades) ->
-    LockFile = lock_filename(),
-    case file:open(LockFile, [write, exclusive]) of
-        {ok, Lock} ->
-            ok = file:close(Lock),
+    LockFile = lock_filename(dir()),
+    case rabbit_misc:lock_file(LockFile) of
+        ok ->
+            BackupDir = dir() ++ "-upgrade-backup",
             info("Upgrades: ~w to apply~n", [length(Upgrades)]),
-            [apply_upgrade(Upgrade) || Upgrade <- Upgrades],
-            info("Upgrades: All applied~n", []),
-            ok = write_version(),
-            ok = file:delete(LockFile);
+            case rabbit_mnesia:copy_db(BackupDir) of
+                ok ->
+                    %% We need to make the backup after creating the
+                    %% lock file so that it protects us from trying to
+                    %% overwrite the backup. Unfortunately this means
+                    %% the lock file exists in the backup too, which
+                    %% is not intuitive. Remove it.
+                    ok = file:delete(lock_filename(BackupDir)),
+                    info("Upgrades: Mnesia dir backed up to ~p~n", [BackupDir]),
+                    [apply_upgrade(Upgrade) || Upgrade <- Upgrades],
+                    info("Upgrades: All upgrades applied successfully~n", []),
+                    ok = write_version(),
+                    ok = rabbit_misc:recursive_delete([BackupDir]),
+                    info("Upgrades: Mnesia backup removed~n", []),
+                    ok = file:delete(LockFile);
+                {error, E} ->
+                    %% If we can't backup, the upgrade hasn't started
+                    %% hence we don't need the lockfile since the real
+                    %% mnesia dir is the good one.
+                    ok = file:delete(LockFile),
+                    throw({could_not_back_up_mnesia_dir, E})
+            end;
         {error, eexist} ->
-            exit(previous_upgrade_failed);
-        {error, _} = Error ->
-            exit(Error)
+            throw({error, previous_upgrade_failed})
     end.
 
 apply_upgrade({M, F}) ->
@@ -141,16 +163,12 @@ apply_upgrade({M, F}) ->
 
 %% -------------------------------------------------------------------
 
-schema_filename() ->
-    filename:join(dir(), ?VERSION_FILENAME).
+dir() -> rabbit_mnesia:dir().
 
-lock_filename() ->
-    filename:join(dir(), ?LOCK_FILENAME).
+schema_filename() -> filename:join(dir(), ?VERSION_FILENAME).
+
+lock_filename(Dir) -> filename:join(Dir, ?LOCK_FILENAME).
 
 %% NB: we cannot use rabbit_log here since it may not have been
 %% started yet
-info(Msg, Args) ->
-    error_logger:info_msg(Msg, Args).
-
-dir() ->
-    rabbit_mnesia:dir().
+info(Msg, Args) -> error_logger:info_msg(Msg, Args).
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 69d62fde..0db51165 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -42,7 +42,7 @@
 -export([start/1, stop/0]).
 
 %% exported for testing only
--export([start_msg_store/2, stop_msg_store/0]).
+-export([start_msg_store/2, stop_msg_store/0, init/5]).
 
 %%----------------------------------------------------------------------------
 %% Definitions:
@@ -89,12 +89,14 @@
 %%
 %% The duration indicated to us by the memory_monitor is used to
 %% calculate, given our current ingress and egress rates, how many
-%% messages we should hold in RAM. When we need to push alphas to
-%% betas or betas to gammas, we favour writing out messages that are
-%% further from the head of the queue. This minimises writes to disk,
-%% as the messages closer to the tail of the queue stay in the queue
-%% for longer, thus do not need to be replaced as quickly by sending
-%% other messages to disk.
+%% messages we should hold in RAM. We track the ingress and egress
+%% rates for both messages and pending acks and rates for both are
+%% considered when calculating the number of messages to hold in
+%% RAM. When we need to push alphas to betas or betas to gammas, we
+%% favour writing out messages that are further from the head of the
+%% queue. This minimises writes to disk, as the messages closer to the
+%% tail of the queue stay in the queue for longer, thus do not need to
+%% be replaced as quickly by sending other messages to disk.
 %%
 %% Whilst messages are pushed to disk and forgotten from RAM as soon
 %% as requested by a new setting of the queue RAM duration, the
@@ -156,7 +158,7 @@
 %% The conversion from alphas to betas is also chunked, but only to
 %% ensure no more than ?IO_BATCH_SIZE alphas are converted to betas at
 %% any one time. This further smooths the effects of changes to the
-%% target_ram_msg_count and ensures the queue remains responsive
+%% target_ram_count and ensures the queue remains responsive
 %% even when there is a large amount of IO work to do. The
 %% idle_timeout callback is utilised to ensure that conversions are
 %% done as promptly as possible whilst ensuring the queue remains
@@ -168,6 +170,29 @@
 %% the latter) are both cheap and do require any scanning through qi
 %% segments.
 %%
+%% Pending acks are recorded in memory either as the tuple {SeqId,
+%% Guid, MsgProps} (tuple-form) or as the message itself (message-
+%% form). Acks for persistent messages are always stored in the tuple-
+%% form. Acks for transient messages are also stored in tuple-form if
+%% the message has been sent to disk as part of the memory reduction
+%% process. For transient messages that haven't already been written
+%% to disk, acks are stored in message-form.
+%%
+%% During memory reduction, acks stored in message-form are converted
+%% to tuple-form, and the corresponding messages are pushed out to
+%% disk.
+%%
+%% The order in which alphas are pushed to betas and message-form acks
+%% are pushed to disk is determined dynamically. We always prefer to
+%% push messages for the source (alphas or acks) that is growing the
+%% fastest (with growth measured as avg. ingress - avg. egress). In
+%% each round of memory reduction a chunk of messages at most
+%% ?IO_BATCH_SIZE in size is allocated to be pushed to disk. The
+%% fastest growing source will be reduced by as much of this chunk as
+%% possible. If there is any remaining allocation in the chunk after
+%% the first source has been reduced to zero, the second source will
+%% be reduced by as much of the remaining chunk as possible.
+%%
 %% Notes on Clean Shutdown
 %% (This documents behaviour in variable_queue, queue_index and
 %% msg_store.)
@@ -220,6 +245,8 @@
           q4,
           next_seq_id,
           pending_ack,
+          pending_ack_index,
+          ram_ack_index,
           index_state,
           msg_store_clients,
           on_sync,
@@ -229,14 +256,21 @@
           len,
           persistent_count,
 
-          target_ram_msg_count,
+          target_ram_count,
           ram_msg_count,
           ram_msg_count_prev,
+          ram_ack_count_prev,
           ram_index_count,
           out_counter,
           in_counter,
-          rates
-         }).
+          rates,
+          msgs_on_disk,
+          msg_indices_on_disk,
+          unconfirmed,
+          ack_out_counter,
+          ack_in_counter,
+          ack_rates
+        }).
 
 -record(rates, { egress, ingress, avg_egress, avg_ingress, timestamp }).
 
@@ -299,30 +333,37 @@
                         funs            :: [fun (() -> any())] }).
 
 -type(state() :: #vqstate {
-             q1                   :: queue(),
-             q2                   :: bpqueue:bpqueue(),
-             delta                :: delta(),
-             q3                   :: bpqueue:bpqueue(),
-             q4                   :: queue(),
-             next_seq_id          :: seq_id(),
-             pending_ack          :: dict(),
-             index_state          :: any(),
-             msg_store_clients    :: 'undefined' | {{any(), binary()},
+             q1                    :: queue(),
+             q2                    :: bpqueue:bpqueue(),
+             delta                 :: delta(),
+             q3                    :: bpqueue:bpqueue(),
+             q4                    :: queue(),
+             next_seq_id           :: seq_id(),
+             pending_ack           :: dict(),
+             ram_ack_index         :: gb_tree(),
+             index_state           :: any(),
+             msg_store_clients     :: 'undefined' | {{any(), binary()},
                                                     {any(), binary()}},
-             on_sync              :: sync(),
-             durable              :: boolean(),
-
-             len                  :: non_neg_integer(),
-             persistent_count     :: non_neg_integer(),
-
-             transient_threshold  :: non_neg_integer(),
-             target_ram_msg_count :: non_neg_integer() | 'infinity',
-             ram_msg_count        :: non_neg_integer(),
-             ram_msg_count_prev   :: non_neg_integer(),
-             ram_index_count      :: non_neg_integer(),
-             out_counter          :: non_neg_integer(),
-             in_counter           :: non_neg_integer(),
-             rates                :: rates() }).
+             on_sync               :: sync(),
+             durable               :: boolean(),
+
+             len                   :: non_neg_integer(),
+             persistent_count      :: non_neg_integer(),
+
+             transient_threshold   :: non_neg_integer(),
+             target_ram_count      :: non_neg_integer() | 'infinity',
+             ram_msg_count         :: non_neg_integer(),
+             ram_msg_count_prev    :: non_neg_integer(),
+             ram_index_count       :: non_neg_integer(),
+             out_counter           :: non_neg_integer(),
+             in_counter            :: non_neg_integer(),
+             rates                 :: rates(),
+             msgs_on_disk          :: gb_set(),
+             msg_indices_on_disk   :: gb_set(),
+             unconfirmed           :: gb_set(),
+             ack_out_counter       :: non_neg_integer(),
+             ack_in_counter        :: non_neg_integer(),
+             ack_rates             :: rates() }).
 
 -include("rabbit_backing_queue_spec.hrl").
 
@@ -368,16 +409,23 @@ stop_msg_store() ->
     ok = rabbit_sup:stop_child(?PERSISTENT_MSG_STORE),
     ok = rabbit_sup:stop_child(?TRANSIENT_MSG_STORE).
 
-init(QueueName, IsDurable, false) ->
-    IndexState = rabbit_queue_index:init(QueueName),
+init(QueueName, IsDurable, Recover) ->
+    Self = self(),
+    init(QueueName, IsDurable, Recover,
+         fun (Guids) -> msgs_written_to_disk(Self, Guids) end,
+         fun (Guids) -> msg_indices_written_to_disk(Self, Guids) end).
+
+init(QueueName, IsDurable, false, MsgOnDiskFun, MsgIdxOnDiskFun) ->
+    IndexState = rabbit_queue_index:init(QueueName, MsgIdxOnDiskFun),
     init(IsDurable, IndexState, 0, [],
          case IsDurable of
-             true  -> msg_store_client_init(?PERSISTENT_MSG_STORE);
+             true  -> msg_store_client_init(?PERSISTENT_MSG_STORE,
+                                            MsgOnDiskFun);
              false -> undefined
          end,
-         msg_store_client_init(?TRANSIENT_MSG_STORE));
+         msg_store_client_init(?TRANSIENT_MSG_STORE, undefined));
 
-init(QueueName, true, true) ->
+init(QueueName, true, true, MsgOnDiskFun, MsgIdxOnDiskFun) ->
     Terms = rabbit_queue_index:shutdown_terms(QueueName),
     {PRef, TRef, Terms1} =
         case [persistent_ref, transient_ref] -- proplists:get_keys(Terms) of
@@ -387,16 +435,17 @@ init(QueueName, true, true) ->
             _  -> {rabbit_guid:guid(), rabbit_guid:guid(), []}
         end,
     PersistentClient = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE,
-                                                    PRef),
+                                                    PRef, MsgOnDiskFun),
     TransientClient  = rabbit_msg_store:client_init(?TRANSIENT_MSG_STORE,
-                                                    TRef),
+                                                    TRef, undefined),
     {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),
+          end,
+          MsgIdxOnDiskFun),
     init(true, IndexState, DeltaCount, Terms1,
          PersistentClient, TransientClient).
 
@@ -472,26 +521,30 @@ publish(Msg, MsgProps, State) ->
 
 publish_delivered(false, _Msg, _MsgProps, State = #vqstate { len = 0 }) ->
     {blank_ack, a(State)};
-publish_delivered(true, Msg = #basic_message { is_persistent = IsPersistent },
-                  MsgProps,
+publish_delivered(true, Msg = #basic_message { is_persistent = IsPersistent,
+                                               guid = Guid },
+                  MsgProps = #message_properties {
+                    needs_confirming = NeedsConfirming },
                   State = #vqstate { len              = 0,
                                      next_seq_id      = SeqId,
                                      out_counter      = OutCount,
                                      in_counter       = InCount,
                                      persistent_count = PCount,
-                                     pending_ack      = PA,
-                                     durable          = IsDurable }) ->
+                                     durable          = IsDurable,
+                                     unconfirmed      = Unconfirmed }) ->
     IsPersistent1 = IsDurable andalso IsPersistent,
     MsgStatus = (msg_status(IsPersistent1, SeqId, Msg, MsgProps))
         #msg_status { is_delivered = true },
     {MsgStatus1, State1} = maybe_write_to_disk(false, false, MsgStatus, State),
-    PA1 = record_pending_ack(m(MsgStatus1), PA),
+    State2 = record_pending_ack(m(MsgStatus1), State1),
     PCount1 = PCount + one_if(IsPersistent1),
-    {SeqId, a(State1 #vqstate { next_seq_id      = SeqId    + 1,
-                                out_counter      = OutCount + 1,
-                                in_counter       = InCount  + 1,
-                                persistent_count = PCount1,
-                                pending_ack      = PA1 })}.
+    Unconfirmed1 = gb_sets_maybe_insert(NeedsConfirming, Guid, Unconfirmed),
+    {SeqId, a(reduce_memory_use(
+                State2 #vqstate { next_seq_id      = SeqId    + 1,
+                                  out_counter      = OutCount + 1,
+                                  in_counter       = InCount  + 1,
+                                  persistent_count = PCount1,
+                                  unconfirmed      = Unconfirmed1 }))}.
 
 dropwhile(Pred, State) ->
     {_OkOrEmpty, State1} = dropwhile1(Pred, State),
@@ -561,8 +614,7 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
                                  index_state       = IndexState,
                                  msg_store_clients = MSCState,
                                  len               = Len,
-                                 persistent_count  = PCount,
-                                 pending_ack       = PA }) ->
+                                 persistent_count  = PCount }) ->
     %% 1. Mark it delivered if necessary
     IndexState1 = maybe_write_delivered(
                     IndexOnDisk andalso not IsDelivered,
@@ -582,12 +634,12 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
         end,
 
     %% 3. If an ack is required, add something sensible to PA
-    {AckTag, PA1} = case AckRequired of
-                        true  -> PA2 = record_pending_ack(
-                                         MsgStatus #msg_status {
-                                           is_delivered = true }, PA),
-                                 {SeqId, PA2};
-                        false -> {blank_ack, PA}
+    {AckTag, State1} = case AckRequired of
+                        true  -> StateN = record_pending_ack(
+                                            MsgStatus #msg_status {
+                                              is_delivered = true }, State),
+                                 {SeqId, StateN};
+                        false -> {blank_ack, State}
                     end,
 
     PCount1 = PCount - one_if(IsPersistent andalso not AckRequired),
@@ -595,17 +647,22 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
     RamMsgCount1 = RamMsgCount - one_if(Msg =/= undefined),
 
     {{Msg, IsDelivered, AckTag, Len1},
-     a(State #vqstate { ram_msg_count    = RamMsgCount1,
-                        out_counter      = OutCount + 1,
-                        index_state      = IndexState2,
-                        len              = Len1,
-                        persistent_count = PCount1,
-                        pending_ack      = PA1 })}.
+     a(State1 #vqstate { ram_msg_count    = RamMsgCount1,
+                         out_counter      = OutCount + 1,
+                         index_state      = IndexState2,
+                         len              = Len1,
+                         persistent_count = PCount1 })}.
 
 ack(AckTags, State) ->
-    a(ack(fun msg_store_remove/3,
-          fun (_AckEntry, State1) -> State1 end,
-          AckTags, State)).
+    {Guids, State1} =
+        ack(fun msg_store_remove/3,
+            fun ({_IsPersistent, Guid, _MsgProps}, State1) ->
+                    remove_confirms(gb_sets:singleton(Guid), State1);
+                (#msg_status{msg = #basic_message{guid = Guid}}, State1) ->
+                    remove_confirms(gb_sets:singleton(Guid), State1)
+            end,
+            AckTags, State),
+    {Guids, a(State1)}.
 
 tx_publish(Txn, Msg = #basic_message { is_persistent = IsPersistent }, MsgProps,
            State = #vqstate { durable           = IsDurable,
@@ -655,7 +712,7 @@ tx_commit(Txn, Fun, MsgPropsFun,
        end)}.
 
 requeue(AckTags, MsgPropsFun, State) ->
-    a(reduce_memory_use(
+    {_Guids, State1} =
         ack(fun msg_store_release/3,
             fun (#msg_status { msg = Msg, msg_props = MsgProps }, State1) ->
                     {_SeqId, State2} = publish(Msg, MsgPropsFun(MsgProps),
@@ -670,48 +727,69 @@ requeue(AckTags, MsgPropsFun, State) ->
                                                true, true, State2),
                     State3
             end,
-            AckTags, State))).
+            AckTags, State),
+    a(reduce_memory_use(State1)).
 
 len(#vqstate { len = Len }) -> Len.
 
 is_empty(State) -> 0 == len(State).
 
-set_ram_duration_target(DurationTarget,
-                        State = #vqstate {
-                          rates = #rates { avg_egress  = AvgEgressRate,
-                                           avg_ingress = AvgIngressRate },
-                          target_ram_msg_count = TargetRamMsgCount }) ->
-    Rate = AvgEgressRate + AvgIngressRate,
-    TargetRamMsgCount1 =
+set_ram_duration_target(
+  DurationTarget, State = #vqstate {
+                    rates     = #rates { avg_egress  = AvgEgressRate,
+                                         avg_ingress = AvgIngressRate },
+                    ack_rates = #rates { avg_egress  = AvgAckEgressRate,
+                                         avg_ingress = AvgAckIngressRate },
+                    target_ram_count = TargetRamCount }) ->
+    Rate =
+        AvgEgressRate + AvgIngressRate + AvgAckEgressRate + AvgAckIngressRate,
+    TargetRamCount1 =
         case DurationTarget of
             infinity  -> infinity;
             _         -> trunc(DurationTarget * Rate) %% msgs = sec * msgs/sec
         end,
-    State1 = State #vqstate { target_ram_msg_count = TargetRamMsgCount1 },
-    a(case TargetRamMsgCount1 == infinity orelse
-          (TargetRamMsgCount =/= infinity andalso
-           TargetRamMsgCount1 >= TargetRamMsgCount) of
+    State1 = State #vqstate { target_ram_count = TargetRamCount1 },
+    a(case TargetRamCount1 == infinity orelse
+          (TargetRamCount =/= infinity andalso
+           TargetRamCount1 >= TargetRamCount) of
           true  -> State1;
           false -> reduce_memory_use(State1)
       end).
 
 ram_duration(State = #vqstate {
-               rates              = #rates { egress    = Egress,
-                                             ingress   = Ingress,
-                                             timestamp = Timestamp } = Rates,
+               rates              = #rates { timestamp = Timestamp,
+                                             egress    = Egress,
+                                             ingress   = Ingress } = Rates,
+               ack_rates          = #rates { timestamp = AckTimestamp,
+                                             egress    = AckEgress,
+                                             ingress   = AckIngress } = ARates,
                in_counter         = InCount,
                out_counter        = OutCount,
+               ack_in_counter     = AckInCount,
+               ack_out_counter    = AckOutCount,
                ram_msg_count      = RamMsgCount,
-               ram_msg_count_prev = RamMsgCountPrev }) ->
+               ram_msg_count_prev = RamMsgCountPrev,
+               ram_ack_index      = RamAckIndex,
+               ram_ack_count_prev = RamAckCountPrev }) ->
     Now = now(),
     {AvgEgressRate,   Egress1} = update_rate(Now, Timestamp, OutCount, Egress),
     {AvgIngressRate, Ingress1} = update_rate(Now, Timestamp, InCount, Ingress),
 
-    Duration = %% msgs / (msgs/sec) == sec
-        case AvgEgressRate == 0 andalso AvgIngressRate == 0 of
+    {AvgAckEgressRate,   AckEgress1} =
+        update_rate(Now, AckTimestamp, AckOutCount, AckEgress),
+    {AvgAckIngressRate, AckIngress1} =
+        update_rate(Now, AckTimestamp, AckInCount, AckIngress),
+
+    RamAckCount = gb_trees:size(RamAckIndex),
+
+    Duration = %% msgs+acks / (msgs+acks/sec) == sec
+        case AvgEgressRate == 0 andalso AvgIngressRate == 0 andalso
+             AvgAckEgressRate == 0 andalso AvgAckIngressRate == 0 of
             true  -> infinity;
-            false -> (RamMsgCountPrev + RamMsgCount) /
-                         (2 * (AvgEgressRate + AvgIngressRate))
+            false -> (RamMsgCountPrev + RamMsgCount +
+                          RamAckCount + RamAckCountPrev) /
+                         (4 * (AvgEgressRate + AvgIngressRate +
+                                   AvgAckEgressRate + AvgAckIngressRate))
         end,
 
     {Duration, State #vqstate {
@@ -721,14 +799,24 @@ ram_duration(State = #vqstate {
                                         avg_egress  = AvgEgressRate,
                                         avg_ingress = AvgIngressRate,
                                         timestamp   = Now },
+                 ack_rates          = ARates #rates {
+                                        egress      = AckEgress1,
+                                        ingress     = AckIngress1,
+                                        avg_egress  = AvgAckEgressRate,
+                                        avg_ingress = AvgAckIngressRate,
+                                        timestamp   = Now },
                  in_counter         = 0,
                  out_counter        = 0,
-                 ram_msg_count_prev = RamMsgCount }}.
+                 ack_in_counter     = 0,
+                 ack_out_counter    = 0,
+                 ram_msg_count_prev = RamMsgCount,
+                 ram_ack_count_prev = RamAckCount }}.
 
 needs_idle_timeout(State = #vqstate { on_sync = ?BLANK_SYNC }) ->
-    {Res, _State} = reduce_memory_use(fun (_Quota, State1) -> State1 end,
+    {Res, _State} = reduce_memory_use(fun (_Quota, State1) -> {0, State1} end,
                                       fun (_Quota, State1) -> State1 end,
                                       fun (State1)         -> State1 end,
+                                      fun (_Quota, State1) -> {0, State1} end,
                                       State),
     Res;
 needs_idle_timeout(_State) ->
@@ -739,33 +827,39 @@ idle_timeout(State) -> a(reduce_memory_use(tx_commit_index(State))).
 handle_pre_hibernate(State = #vqstate { index_state = IndexState }) ->
     State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }.
 
-status(#vqstate { q1 = Q1, q2 = Q2, delta = Delta, q3 = Q3, q4 = Q4,
-                  len                  = Len,
-                  pending_ack          = PA,
-                  on_sync              = #sync { funs = From },
-                  target_ram_msg_count = TargetRamMsgCount,
-                  ram_msg_count        = RamMsgCount,
-                  ram_index_count      = RamIndexCount,
-                  next_seq_id          = NextSeqId,
-                  persistent_count     = PersistentCount,
-                  rates                = #rates {
-                    avg_egress  = AvgEgressRate,
-                    avg_ingress = AvgIngressRate } }) ->
-    [ {q1                   , queue:len(Q1)},
-      {q2                   , bpqueue:len(Q2)},
-      {delta                , Delta},
-      {q3                   , bpqueue:len(Q3)},
-      {q4                   , queue:len(Q4)},
-      {len                  , Len},
-      {pending_acks         , dict:size(PA)},
-      {outstanding_txns     , length(From)},
-      {target_ram_msg_count , TargetRamMsgCount},
-      {ram_msg_count        , RamMsgCount},
-      {ram_index_count      , RamIndexCount},
-      {next_seq_id          , NextSeqId},
-      {persistent_count     , PersistentCount},
-      {avg_egress_rate      , AvgEgressRate},
-      {avg_ingress_rate     , AvgIngressRate} ].
+status(#vqstate {
+          q1 = Q1, q2 = Q2, delta = Delta, q3 = Q3, q4 = Q4,
+          len              = Len,
+          pending_ack      = PA,
+          ram_ack_index    = RAI,
+          on_sync          = #sync { funs = From },
+          target_ram_count = TargetRamCount,
+          ram_msg_count    = RamMsgCount,
+          ram_index_count  = RamIndexCount,
+          next_seq_id      = NextSeqId,
+          persistent_count = PersistentCount,
+          rates            = #rates { avg_egress  = AvgEgressRate,
+                                      avg_ingress = AvgIngressRate },
+          ack_rates        = #rates { avg_egress  = AvgAckEgressRate,
+                                      avg_ingress = AvgAckIngressRate } }) ->
+    [ {q1                  , queue:len(Q1)},
+      {q2                  , bpqueue:len(Q2)},
+      {delta               , Delta},
+      {q3                  , bpqueue:len(Q3)},
+      {q4                  , queue:len(Q4)},
+      {len                 , Len},
+      {pending_acks        , dict:size(PA)},
+      {outstanding_txns    , length(From)},
+      {target_ram_count    , TargetRamCount},
+      {ram_msg_count       , RamMsgCount},
+      {ram_ack_count       , gb_trees:size(RAI)},
+      {ram_index_count     , RamIndexCount},
+      {next_seq_id         , NextSeqId},
+      {persistent_count    , PersistentCount},
+      {avg_ingress_rate    , AvgIngressRate},
+      {avg_egress_rate     , AvgEgressRate},
+      {avg_ack_ingress_rate, AvgAckIngressRate},
+      {avg_ack_egress_rate , AvgAckEgressRate} ].
 
 %%----------------------------------------------------------------------------
 %% Minor helpers
@@ -811,6 +905,10 @@ one_if(false) -> 0.
 cons_if(true,   E, L) -> [E | L];
 cons_if(false, _E, L) -> L.
 
+gb_sets_maybe_insert(false, _Val, Set) -> Set;
+%% when requeueing, we re-add a guid to the unconfimred set
+gb_sets_maybe_insert(true,  Val,  Set) -> gb_sets:add(Val, Set).
+
 msg_status(IsPersistent, SeqId, Msg = #basic_message { guid = Guid },
            MsgProps) ->
     #msg_status { seq_id = SeqId, guid = Guid, msg = Msg,
@@ -832,8 +930,8 @@ with_immutable_msg_store_state(MSCState, IsPersistent, Fun) ->
                                            end),
     Res.
 
-msg_store_client_init(MsgStore) ->
-    rabbit_msg_store:client_init(MsgStore, rabbit_guid:guid()).
+msg_store_client_init(MsgStore, MsgOnDiskFun) ->
+    rabbit_msg_store:client_init(MsgStore, rabbit_guid:guid(), MsgOnDiskFun).
 
 msg_store_write(MSCState, IsPersistent, Guid, Msg) ->
     with_immutable_msg_store_state(
@@ -955,35 +1053,46 @@ init(IsDurable, IndexState, DeltaCount, Terms,
             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,
-
-      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 } },
+      q1                  = queue:new(),
+      q2                  = bpqueue:new(),
+      delta               = Delta,
+      q3                  = bpqueue:new(),
+      q4                  = queue:new(),
+      next_seq_id         = NextSeqId,
+      pending_ack         = dict:new(),
+      ram_ack_index       = gb_trees:empty(),
+      index_state         = IndexState1,
+      msg_store_clients   = {PersistentClient, TransientClient},
+      on_sync             = ?BLANK_SYNC,
+      durable             = IsDurable,
+      transient_threshold = NextSeqId,
+
+      len                 = DeltaCount1,
+      persistent_count    = DeltaCount1,
+
+      target_ram_count    = infinity,
+      ram_msg_count       = 0,
+      ram_msg_count_prev  = 0,
+      ram_ack_count_prev  = 0,
+      ram_index_count     = 0,
+      out_counter         = 0,
+      in_counter          = 0,
+      rates               = blank_rate(Now, DeltaCount1),
+      msgs_on_disk        = gb_sets:new(),
+      msg_indices_on_disk = gb_sets:new(),
+      unconfirmed         = gb_sets:new(),
+      ack_out_counter     = 0,
+      ack_in_counter      = 0,
+      ack_rates           = blank_rate(Now, 0) },
     a(maybe_deltas_to_betas(State)).
 
+blank_rate(Timestamp, IngressLength) ->
+    #rates { egress      = {Timestamp, 0},
+             ingress     = {Timestamp, IngressLength},
+             avg_egress  = 0.0,
+             avg_ingress = 0.0,
+             timestamp   = Timestamp }.
+
 msg_store_callback(PersistentGuids, Pubs, AckTags, Fun, MsgPropsFun) ->
     Self = self(),
     F = fun () -> rabbit_amqqueue:maybe_run_queue_via_backing_queue(
@@ -1000,7 +1109,7 @@ msg_store_callback(PersistentGuids, Pubs, AckTags, Fun, MsgPropsFun) ->
     end.
 
 remove_persistent_messages(Guids) ->
-    PersistentClient = msg_store_client_init(?PERSISTENT_MSG_STORE),
+    PersistentClient = msg_store_client_init(?PERSISTENT_MSG_STORE, undefined),
     ok = rabbit_msg_store:remove(Guids, PersistentClient),
     rabbit_msg_store:client_delete_and_terminate(PersistentClient).
 
@@ -1051,6 +1160,7 @@ tx_commit_index(State = #vqstate { on_sync = #sync {
                                    durable = IsDurable }) ->
     PAcks = lists:append(SPAcks),
     Acks  = lists:append(SAcks),
+    {_Guids, NewState} = ack(Acks, State),
     Pubs  = [{Msg, Fun(MsgProps)} || {Fun, PubsN}    <- lists:reverse(SPubs),
                                      {Msg, MsgProps} <- lists:reverse(PubsN)],
     {SeqIds, State1 = #vqstate { index_state = IndexState }} =
@@ -1062,7 +1172,7 @@ tx_commit_index(State = #vqstate { on_sync = #sync {
                   {SeqId, State3} =
                       publish(Msg, MsgProps, false, IsPersistent1, State2),
                   {cons_if(IsPersistent1, SeqId, SeqIdsAcc), State3}
-          end, {PAcks, ack(Acks, State)}, Pubs),
+          end, {PAcks, NewState}, Pubs),
     IndexState1 = rabbit_queue_index:sync(SeqIds, IndexState),
     [ Fun() || Fun <- lists:reverse(SFuns) ],
     reduce_memory_use(
@@ -1116,15 +1226,17 @@ sum_guids_by_store_to_len(LensByStore, GuidsByStore) ->
 %% Internal gubbins for publishing
 %%----------------------------------------------------------------------------
 
-publish(Msg = #basic_message { is_persistent = IsPersistent },
-        MsgProps, IsDelivered, MsgOnDisk,
+publish(Msg = #basic_message { is_persistent = IsPersistent, guid = Guid },
+        MsgProps = #message_properties { needs_confirming = NeedsConfirming },
+        IsDelivered, MsgOnDisk,
         State = #vqstate { q1 = Q1, q3 = Q3, q4 = Q4,
                            next_seq_id      = SeqId,
                            len              = Len,
                            in_counter       = InCount,
                            persistent_count = PCount,
                            durable          = IsDurable,
-                           ram_msg_count    = RamMsgCount }) ->
+                           ram_msg_count    = RamMsgCount,
+                           unconfirmed      = Unconfirmed }) ->
     IsPersistent1 = IsDurable andalso IsPersistent,
     MsgStatus = (msg_status(IsPersistent1, SeqId, Msg, MsgProps))
         #msg_status { is_delivered = IsDelivered, msg_on_disk = MsgOnDisk},
@@ -1134,11 +1246,13 @@ publish(Msg = #basic_message { is_persistent = IsPersistent },
                  true  -> State1 #vqstate { q4 = queue:in(m(MsgStatus1), Q4) }
              end,
     PCount1 = PCount + one_if(IsPersistent1),
+    Unconfirmed1 = gb_sets_maybe_insert(NeedsConfirming, Guid, Unconfirmed),
     {SeqId, State2 #vqstate { next_seq_id      = SeqId   + 1,
                               len              = Len     + 1,
                               in_counter       = InCount + 1,
                               persistent_count = PCount1,
-                              ram_msg_count    = RamMsgCount + 1}}.
+                              ram_msg_count    = RamMsgCount + 1,
+                              unconfirmed      = Unconfirmed1 }}.
 
 maybe_write_msg_to_disk(_Force, MsgStatus = #msg_status {
                                   msg_on_disk = true }, _MSCState) ->
@@ -1191,12 +1305,21 @@ record_pending_ack(#msg_status { seq_id        = SeqId,
                                  guid          = Guid,
                                  is_persistent = IsPersistent,
                                  msg_on_disk   = MsgOnDisk,
-                                 msg_props     = MsgProps } = MsgStatus, PA) ->
-    AckEntry = case MsgOnDisk of
-                   true  -> {IsPersistent, Guid, MsgProps};
-                   false -> MsgStatus
-               end,
-    dict:store(SeqId, AckEntry, PA).
+                                 msg_props     = MsgProps } = MsgStatus,
+                   State = #vqstate { pending_ack     = PA,
+                                      ram_ack_index   = RAI,
+                                      ack_in_counter  = AckInCount}) ->
+    {AckEntry, RAI1} =
+        case MsgOnDisk of
+            true ->
+                {{IsPersistent, Guid, MsgProps}, RAI};
+            false ->
+                {MsgStatus, gb_trees:insert(SeqId, Guid, RAI)}
+        end,
+    PA1 = dict:store(SeqId, AckEntry, PA),
+    State #vqstate { pending_ack    = PA1,
+                     ram_ack_index  = RAI1,
+                     ack_in_counter = AckInCount + 1}.
 
 remove_pending_ack(KeepPersistent,
                    State = #vqstate { pending_ack       = PA,
@@ -1204,7 +1327,8 @@ remove_pending_ack(KeepPersistent,
                                       msg_store_clients = MSCState }) ->
     {SeqIds, GuidsByStore} = dict:fold(fun accumulate_ack/3,
                                        {[], orddict:new()}, PA),
-    State1 = State #vqstate { pending_ack = dict:new() },
+    State1 = State #vqstate { pending_ack   = dict:new(),
+                              ram_ack_index = gb_trees:empty() },
     case KeepPersistent of
         true  -> case orddict:find(false, GuidsByStore) of
                      error       -> State1;
@@ -1221,27 +1345,35 @@ remove_pending_ack(KeepPersistent,
     end.
 
 ack(_MsgStoreFun, _Fun, [], State) ->
-    State;
+    {[], State};
 ack(MsgStoreFun, Fun, AckTags, State) ->
     {{SeqIds, GuidsByStore},
      State1 = #vqstate { index_state       = IndexState,
                          msg_store_clients = MSCState,
-                         persistent_count  = PCount }} =
+                         persistent_count  = PCount,
+                         ack_out_counter   = AckOutCount }} =
         lists:foldl(
-          fun (SeqId, {Acc, State2 = #vqstate { pending_ack = PA }}) ->
+          fun (SeqId, {Acc, State2 = #vqstate { pending_ack   = PA,
+                                                ram_ack_index = RAI }}) ->
                   AckEntry = dict:fetch(SeqId, PA),
                   {accumulate_ack(SeqId, AckEntry, Acc),
                    Fun(AckEntry, State2 #vqstate {
-                                   pending_ack = dict:erase(SeqId, PA) })}
+                                   pending_ack   = dict:erase(SeqId, PA),
+                                   ram_ack_index =
+                                       gb_trees:delete_any(SeqId, RAI)})}
           end, {{[], orddict:new()}, State}, AckTags),
     IndexState1 = rabbit_queue_index:ack(SeqIds, IndexState),
-    ok = orddict:fold(fun (IsPersistent, Guids, ok) ->
-                              MsgStoreFun(MSCState, IsPersistent, Guids)
-                      end, ok, GuidsByStore),
+    AckdGuids = lists:concat(
+                  orddict:fold(
+                    fun (IsPersistent, Guids, Gs) ->
+                            MsgStoreFun(MSCState, IsPersistent, Guids),
+                            [Guids | Gs]
+                    end, [], GuidsByStore)),
     PCount1 = PCount - find_persistent_count(sum_guids_by_store_to_len(
                                                orddict:new(), GuidsByStore)),
-    State1 #vqstate { index_state      = IndexState1,
-                      persistent_count = PCount1 }.
+    {AckdGuids, State1 #vqstate { index_state      = IndexState1,
+                      persistent_count = PCount1,
+                      ack_out_counter  = AckOutCount + length(AckTags) }}.
 
 accumulate_ack(_SeqId, #msg_status { is_persistent = false, %% ASSERTIONS
                                      msg_on_disk   = false,
@@ -1258,6 +1390,44 @@ find_persistent_count(LensByStore) ->
     end.
 
 %%----------------------------------------------------------------------------
+%% Internal plumbing for confirms (aka publisher acks)
+%%----------------------------------------------------------------------------
+
+remove_confirms(GuidSet, State = #vqstate { msgs_on_disk        = MOD,
+                                            msg_indices_on_disk = MIOD,
+                                            unconfirmed         = UC }) ->
+    State #vqstate { msgs_on_disk        = gb_sets:difference(MOD,  GuidSet),
+                     msg_indices_on_disk = gb_sets:difference(MIOD, GuidSet),
+                     unconfirmed         = gb_sets:difference(UC,   GuidSet) }.
+
+msgs_confirmed(GuidSet, State) ->
+    {{confirm, gb_sets:to_list(GuidSet)}, remove_confirms(GuidSet, State)}.
+
+msgs_written_to_disk(QPid, GuidSet) ->
+    rabbit_amqqueue:maybe_run_queue_via_backing_queue_async(
+      QPid, fun(State = #vqstate { msgs_on_disk        = MOD,
+                                   msg_indices_on_disk = MIOD,
+                                   unconfirmed         = UC }) ->
+                    msgs_confirmed(gb_sets:intersection(GuidSet, MIOD),
+                                   State #vqstate {
+                                     msgs_on_disk =
+                                         gb_sets:intersection(
+                                           gb_sets:union(MOD, GuidSet), UC) })
+            end).
+
+msg_indices_written_to_disk(QPid, GuidSet) ->
+    rabbit_amqqueue:maybe_run_queue_via_backing_queue_async(
+      QPid, fun(State = #vqstate { msgs_on_disk        = MOD,
+                                   msg_indices_on_disk = MIOD,
+                                   unconfirmed         = UC }) ->
+                    msgs_confirmed(gb_sets:intersection(GuidSet, MOD),
+                                   State #vqstate {
+                                     msg_indices_on_disk =
+                                         gb_sets:intersection(
+                                           gb_sets:union(MIOD, GuidSet), UC) })
+            end).
+
+%%----------------------------------------------------------------------------
 %% Phase changes
 %%----------------------------------------------------------------------------
 
@@ -1270,7 +1440,7 @@ find_persistent_count(LensByStore) ->
 %% though the conversion function for that is called as necessary. The
 %% reason is twofold. Firstly, this is safe because the conversion is
 %% only ever necessary just after a transition to a
-%% target_ram_msg_count of zero or after an incremental alpha->beta
+%% target_ram_count of zero or after an incremental alpha->beta
 %% conversion. In the former case the conversion is performed straight
 %% away (i.e. any betas present at the time are converted to deltas),
 %% and in the latter case the need for a conversion is flagged up
@@ -1280,26 +1450,77 @@ find_persistent_count(LensByStore) ->
 %% one segment's worth of messages in q3 - and thus would risk
 %% perpetually reporting the need for a conversion when no such
 %% conversion is needed. That in turn could cause an infinite loop.
-reduce_memory_use(AlphaBetaFun, BetaGammaFun, BetaDeltaFun, State) ->
-    {Reduce, State1} = case chunk_size(State #vqstate.ram_msg_count,
-                                       State #vqstate.target_ram_msg_count) of
-                           0  -> {false, State};
-                           S1 -> {true, AlphaBetaFun(S1, State)}
-                       end,
-    case State1 #vqstate.target_ram_msg_count of
-        infinity -> {Reduce, State1};
-        0        -> {Reduce, BetaDeltaFun(State1)};
-        _        -> case chunk_size(State1 #vqstate.ram_index_count,
-                                   permitted_ram_index_count(State1)) of
-                        ?IO_BATCH_SIZE = S2 -> {true, BetaGammaFun(S2, State1)};
-                        _                   -> {Reduce, State1}
-                    end
+reduce_memory_use(_AlphaBetaFun, _BetaGammaFun, _BetaDeltaFun, _AckFun,
+                  State = #vqstate {target_ram_count = infinity}) ->
+    {false, State};
+reduce_memory_use(AlphaBetaFun, BetaGammaFun, BetaDeltaFun, AckFun,
+                  State = #vqstate {
+                    ram_ack_index    = RamAckIndex,
+                    ram_msg_count    = RamMsgCount,
+                    target_ram_count = TargetRamCount,
+                    rates            = #rates { avg_ingress = AvgIngress,
+                                                avg_egress  = AvgEgress },
+                    ack_rates        = #rates { avg_ingress = AvgAckIngress,
+                                                avg_egress  = AvgAckEgress }
+                   }) ->
+
+    {Reduce, State1} =
+        case chunk_size(RamMsgCount + gb_trees:size(RamAckIndex),
+                        TargetRamCount) of
+            0  -> {false, State};
+            %% Reduce memory of pending acks and alphas. The order is
+            %% determined based on which is growing faster. Whichever
+            %% comes second may very well get a quota of 0 if the
+            %% first manages to push out the max number of messages.
+            S1 -> {_, State2} =
+                      lists:foldl(fun (ReduceFun, {QuotaN, StateN}) ->
+                                          ReduceFun(QuotaN, StateN)
+                                  end,
+                                  {S1, State},
+                                  case (AvgAckIngress - AvgAckEgress) >
+                                      (AvgIngress - AvgEgress) of
+                                      true  -> [AckFun, AlphaBetaFun];
+                                      false -> [AlphaBetaFun, AckFun]
+                                  end),
+                  {true, State2}
+        end,
+
+    case State1 #vqstate.target_ram_count of
+        0 -> {Reduce, BetaDeltaFun(State1)};
+        _ -> case chunk_size(State1 #vqstate.ram_index_count,
+                             permitted_ram_index_count(State1)) of
+                 ?IO_BATCH_SIZE = S2 -> {true, BetaGammaFun(S2, State1)};
+                 _                   -> {Reduce, State1}
+             end
     end.
 
+limit_ram_acks(0, State) ->
+    {0, State};
+limit_ram_acks(Quota, State = #vqstate { pending_ack   = PA,
+                                         ram_ack_index = RAI }) ->
+    case gb_trees:is_empty(RAI) of
+        true ->
+            {Quota, State};
+        false ->
+            {SeqId, Guid, RAI1} = gb_trees:take_largest(RAI),
+            MsgStatus = #msg_status {
+              guid          = Guid, %% ASSERTION
+              is_persistent = false, %% ASSERTION
+              msg_props     = MsgProps } = dict:fetch(SeqId, PA),
+            {_, State1} = maybe_write_to_disk(true, false, MsgStatus, State),
+            limit_ram_acks(Quota - 1,
+                           State1 #vqstate {
+                             pending_ack   =
+                                 dict:store(SeqId, {false, Guid, MsgProps}, PA),
+                             ram_ack_index = RAI1 })
+    end.
+
+
 reduce_memory_use(State) ->
     {_, State1} = reduce_memory_use(fun push_alphas_to_betas/2,
                                     fun limit_ram_index/2,
                                     fun push_betas_to_deltas/1,
+                                    fun limit_ram_acks/2,
                                     State),
     State1.
 
@@ -1432,9 +1653,9 @@ maybe_deltas_to_betas(State = #vqstate {
     end.
 
 push_alphas_to_betas(Quota, State) ->
-    { Quota1, State1} = maybe_push_q1_to_betas(Quota,  State),
-    {_Quota2, State2} = maybe_push_q4_to_betas(Quota1, State1),
-    State2.
+    {Quota1, State1} = maybe_push_q1_to_betas(Quota,  State),
+    {Quota2, State2} = maybe_push_q4_to_betas(Quota1, State1),
+    {Quota2, State2}.
 
 maybe_push_q1_to_betas(Quota, State = #vqstate { q1 = Q1 }) ->
     maybe_push_alphas_to_betas(
@@ -1460,10 +1681,11 @@ maybe_push_q4_to_betas(Quota, State = #vqstate { q4 = Q4 }) ->
 
 maybe_push_alphas_to_betas(_Generator, _Consumer, Quota, _Q,
                            State = #vqstate {
-                             ram_msg_count        = RamMsgCount,
-                             target_ram_msg_count = TargetRamMsgCount })
+                             ram_msg_count    = RamMsgCount,
+                             target_ram_count = TargetRamCount })
   when Quota =:= 0 orelse
-       TargetRamMsgCount =:= infinity orelse TargetRamMsgCount >= RamMsgCount ->
+       TargetRamCount =:= infinity orelse
+       TargetRamCount >= RamMsgCount ->
     {Quota, State};
 maybe_push_alphas_to_betas(Generator, Consumer, Quota, Q, State) ->
     case Generator(Q) of
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index 50bca390..068ac186 100644
--- a/src/rabbit_writer.erl
+++ b/src/rabbit_writer.erl
@@ -34,12 +34,11 @@
 -include("rabbit_framing.hrl").
 
 -export([start/5, start_link/5, mainloop/2, mainloop1/2]).
--export([send_command/2, send_command/3, send_command_sync/2,
-         send_command_sync/3, send_command_and_notify/5]).
+-export([send_command/2, send_command/3,
+         send_command_sync/2, send_command_sync/3,
+         send_command_and_notify/4, send_command_and_notify/5]).
 -export([internal_send_command/4, internal_send_command/6]).
 
--import(gen_tcp).
-
 -record(wstate, {sock, channel, frame_max, protocol}).
 
 -define(HIBERNATE_AFTER, 5000).
@@ -66,6 +65,9 @@
 -spec(send_command_sync/3 ::
         (pid(), rabbit_framing:amqp_method_record(), rabbit_types:content())
         -> 'ok').
+-spec(send_command_and_notify/4 ::
+        (pid(), pid(), pid(), rabbit_framing:amqp_method_record())
+        -> 'ok').
 -spec(send_command_and_notify/5 ::
         (pid(), pid(), pid(), rabbit_framing:amqp_method_record(),
          rabbit_types:content())
@@ -130,6 +132,10 @@ handle_message({'$gen_call', From, {send_command_sync, MethodRecord, Content}},
     ok = internal_send_command_async(MethodRecord, Content, State),
     gen_server:reply(From, ok),
     State;
+handle_message({send_command_and_notify, QPid, ChPid, MethodRecord}, State) ->
+    ok = internal_send_command_async(MethodRecord, State),
+    rabbit_amqqueue:notify_sent(QPid, ChPid),
+    State;
 handle_message({send_command_and_notify, QPid, ChPid, MethodRecord, Content},
                State) ->
     ok = internal_send_command_async(MethodRecord, Content, State),
@@ -158,6 +164,10 @@ send_command_sync(W, MethodRecord) ->
 send_command_sync(W, MethodRecord, Content) ->
     call(W, {send_command_sync, MethodRecord, Content}).
 
+send_command_and_notify(W, Q, ChPid, MethodRecord) ->
+    W ! {send_command_and_notify, Q, ChPid, MethodRecord},
+    ok.
+
 send_command_and_notify(W, Q, ChPid, MethodRecord, Content) ->
     W ! {send_command_and_notify, Q, ChPid, MethodRecord, Content},
     ok.
@@ -170,7 +180,7 @@ call(Pid, Msg) ->
 
 %---------------------------------------------------------------------------
 
-assemble_frames(Channel, MethodRecord, Protocol) ->
+assemble_frame(Channel, MethodRecord, Protocol) ->
     ?LOGMESSAGE(out, Channel, MethodRecord, none),
     rabbit_binary_generator:build_simple_method_frame(
       Channel, MethodRecord, Protocol).
@@ -185,17 +195,34 @@ assemble_frames(Channel, MethodRecord, Content, FrameMax, Protocol) ->
                       Channel, Content, FrameMax, Protocol),
     [MethodFrame | ContentFrames].
 
+%% We optimise delivery of small messages. Content-bearing methods
+%% require at least three frames. Small messages always fit into
+%% that. We hand their frames to the Erlang network functions in one
+%% go, which may lead to somewhat more efficient processing in the
+%% runtime and a greater chance of coalescing into fewer TCP packets.
+%%
+%% By contrast, for larger messages, split across many frames, we want
+%% to allow interleaving of frames on different channels. Hence we
+%% hand them to the Erlang network functions one frame at a time.
+send_frames(Fun, Sock, Frames) when length(Frames) =< 3 ->
+    Fun(Sock, Frames);
+send_frames(Fun, Sock, Frames) ->
+    lists:foldl(fun (Frame,     ok) -> Fun(Sock, Frame);
+                    (_Frame, Other) -> Other
+                end, ok, Frames).
+
 tcp_send(Sock, Data) ->
     rabbit_misc:throw_on_error(inet_error,
                                fun () -> rabbit_net:send(Sock, Data) end).
 
 internal_send_command(Sock, Channel, MethodRecord, Protocol) ->
-    ok = tcp_send(Sock, assemble_frames(Channel, MethodRecord, Protocol)).
+    ok = tcp_send(Sock, assemble_frame(Channel, MethodRecord, Protocol)).
 
 internal_send_command(Sock, Channel, MethodRecord, Content, FrameMax,
                       Protocol) ->
-    ok = tcp_send(Sock, assemble_frames(Channel, MethodRecord,
-                                        Content, FrameMax, Protocol)).
+    ok = send_frames(fun tcp_send/2, Sock,
+                     assemble_frames(Channel, MethodRecord,
+                                     Content, FrameMax, Protocol)).
 
 %% gen_tcp:send/2 does a selective receive of {inet_reply, Sock,
 %% Status} to obtain the result. That is bad when it is called from
@@ -219,19 +246,19 @@ internal_send_command_async(MethodRecord,
                             #wstate{sock      = Sock,
                                     channel   = Channel,
                                     protocol  = Protocol}) ->
-    true = port_cmd(Sock, assemble_frames(Channel, MethodRecord, Protocol)),
-    ok.
+    ok = port_cmd(Sock, assemble_frame(Channel, MethodRecord, Protocol)).
 
 internal_send_command_async(MethodRecord, Content,
                             #wstate{sock      = Sock,
                                     channel   = Channel,
                                     frame_max = FrameMax,
                                     protocol  = Protocol}) ->
-    true = port_cmd(Sock, assemble_frames(Channel, MethodRecord,
-                                          Content, FrameMax, Protocol)),
-    ok.
+    ok = send_frames(fun port_cmd/2, Sock,
+                     assemble_frames(Channel, MethodRecord,
+                                     Content, FrameMax, Protocol)).
 
 port_cmd(Sock, Data) ->
-    try rabbit_net:port_command(Sock, Data)
-    catch error:Error -> exit({writer, send_failed, Error})
-    end.
+    true = try rabbit_net:port_command(Sock, Data)
+           catch error:Error -> exit({writer, send_failed, Error})
+           end,
+    ok.