remove transactions from backing queue

...and mostly from amqqueue_process

6 files changed, 60 insertions, 366 deletions

diff --git a/include/rabbit_backing_queue_spec.hrl b/include/rabbit_backing_queue_spec.hrl
index 295d9039..ee102f5e 100644
--- a/include/rabbit_backing_queue_spec.hrl
+++ b/include/rabbit_backing_queue_spec.hrl
@@ -26,12 +26,11 @@
         fun ((rabbit_types:message_properties())
              -> rabbit_types:message_properties())).
 -type(async_callback() :: fun ((atom(), fun ((atom(), state()) -> state())) -> 'ok')).
--type(sync_callback() :: fun ((atom(), fun ((atom(), state()) -> state())) -> 'ok' | 'error')).
 
 -spec(start/1 :: ([rabbit_amqqueue:name()]) -> 'ok').
 -spec(stop/0 :: () -> 'ok').
--spec(init/4 :: (rabbit_types:amqqueue(), attempt_recovery(),
-                 async_callback(), sync_callback()) -> state()).
+-spec(init/3 :: (rabbit_types:amqqueue(), attempt_recovery(),
+                 async_callback()) -> state()).
 -spec(terminate/2 :: (any(), state()) -> state()).
 -spec(delete_and_terminate/2 :: (any(), state()) -> state()).
 -spec(purge/1 :: (state()) -> {purged_msg_count(), state()}).
@@ -51,14 +50,6 @@
 -spec(fetch/2 :: (true,  state()) -> {fetch_result(ack()), state()};
                  (false, state()) -> {fetch_result(undefined), state()}).
 -spec(ack/2 :: ([ack()], state()) -> {[rabbit_guid:guid()], state()}).
--spec(tx_publish/5 :: (rabbit_types:txn(), rabbit_types:basic_message(),
-                       rabbit_types:message_properties(), pid(), state()) ->
-                           state()).
--spec(tx_ack/3 :: (rabbit_types:txn(), [ack()], state()) -> state()).
--spec(tx_rollback/2 :: (rabbit_types:txn(), state()) -> {[ack()], state()}).
--spec(tx_commit/4 ::
-        (rabbit_types:txn(), fun (() -> any()),
-         message_properties_transformer(), state()) -> {[ack()], state()}).
 -spec(requeue/3 :: ([ack()], message_properties_transformer(), state())
                    -> {[rabbit_guid:guid()], state()}).
 -spec(len/1 :: (state()) -> non_neg_integer()).
@@ -71,7 +62,7 @@
 -spec(handle_pre_hibernate/1 :: (state()) -> state()).
 -spec(status/1 :: (state()) -> [{atom(), any()}]).
 -spec(invoke/3 :: (atom(), fun ((atom(), A) -> A), state()) -> state()).
--spec(is_duplicate/3 ::
-        (rabbit_types:txn(), rabbit_types:basic_message(), state()) ->
+-spec(is_duplicate/2 ::
+        (rabbit_types:basic_message(), state()) ->
                              {'false'|'published'|'discarded', state()}).
 -spec(discard/3 :: (rabbit_types:basic_message(), pid(), state()) -> state()).
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index c1fa048d..28fced98 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -62,7 +62,6 @@
              monitor_ref,
              acktags,
              is_limit_active,
-             txn,
              unsent_message_count}).
 
 -define(STATISTICS_KEYS,
@@ -194,13 +193,6 @@ bq_init(BQ, Q, Recover) ->
     BQ:init(Q, Recover,
             fun (Mod, Fun) ->
                     rabbit_amqqueue:run_backing_queue_async(Self, Mod, Fun)
-            end,
-            fun (Mod, Fun) ->
-                    rabbit_misc:with_exit_handler(
-                      fun () -> error end,
-                      fun () ->
-                              rabbit_amqqueue:run_backing_queue(Self, Mod, Fun)
-                      end)
             end).
 
 process_args(State = #q{q = #amqqueue{arguments = Arguments}}) ->
@@ -217,22 +209,14 @@ init_expires(Expires, State) -> ensure_expiry_timer(State#q{expires = Expires}).
 init_ttl(TTL, State) -> drop_expired_messages(State#q{ttl = TTL}).
 
 terminate_shutdown(Fun, State) ->
-    State1 = #q{backing_queue = BQ, backing_queue_state = BQS} =
+    State1 = #q{backing_queue_state = BQS} =
         stop_sync_timer(stop_rate_timer(State)),
     case BQS of
         undefined -> State;
         _         -> ok = rabbit_memory_monitor:deregister(self()),
-                     BQS1 = lists:foldl(
-                              fun (#cr{txn = none}, BQSN) ->
-                                      BQSN;
-                                  (#cr{txn = Txn}, BQSN) ->
-                                      {_AckTags, BQSN1} =
-                                          BQ:tx_rollback(Txn, BQSN),
-                                      BQSN1
-                              end, BQS, all_ch_record()),
                      [emit_consumer_deleted(Ch, CTag)
                       || {Ch, CTag, _} <- consumers(State1)],
-                     State1#q{backing_queue_state = Fun(BQS1)}
+                     State1#q{backing_queue_state = Fun(BQS)}
     end.
 
 reply(Reply, NewState) ->
@@ -343,7 +327,6 @@ ch_record(ChPid) ->
                              monitor_ref          = MonitorRef,
                              acktags              = sets:new(),
                              is_limit_active      = false,
-                             txn                  = none,
                              unsent_message_count = 0},
                      put(Key, C),
                      C;
@@ -355,13 +338,12 @@ store_ch_record(C = #cr{ch_pid = ChPid}) ->
 
 maybe_store_ch_record(C = #cr{consumer_count       = ConsumerCount,
                               acktags              = ChAckTags,
-                              txn                  = Txn,
                               unsent_message_count = UnsentMessageCount}) ->
-    case {sets:size(ChAckTags), ConsumerCount, UnsentMessageCount, Txn} of
-        {0, 0, 0, none} -> ok = erase_ch_record(C),
-                           false;
-        _               -> store_ch_record(C),
-                           true
+    case {sets:size(ChAckTags), ConsumerCount, UnsentMessageCount} of
+        {0, 0, 0} -> ok = erase_ch_record(C),
+                     false;
+        _         -> store_ch_record(C),
+                     true
     end.
 
 erase_ch_record(#cr{ch_pid      = ChPid,
@@ -523,7 +505,7 @@ attempt_delivery(Delivery = #delivery{txn        = none,
         immediately -> rabbit_channel:confirm(ChPid, [MsgSeqNo]);
         _           -> ok
     end,
-    case BQ:is_duplicate(none, Message, BQS) of
+    case BQ:is_duplicate(Message, BQS) of
         {false, BQS1} ->
             PredFun = fun (IsEmpty, _State) -> not IsEmpty end,
             DeliverFun =
@@ -555,24 +537,6 @@ attempt_delivery(Delivery = #delivery{txn        = none,
                             discarded -> false
                         end,
             {Delivered, Confirm, State#q{backing_queue_state = BQS1}}
-    end;
-attempt_delivery(Delivery = #delivery{txn     = Txn,
-                                      sender  = ChPid,
-                                      message = Message},
-                 State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
-    Confirm = should_confirm_message(Delivery, State),
-    case BQ:is_duplicate(Txn, Message, BQS) of
-        {false, BQS1} ->
-            store_ch_record((ch_record(ChPid))#cr{txn = Txn}),
-            BQS2 = BQ:tx_publish(Txn, Message, ?BASE_MESSAGE_PROPERTIES, ChPid,
-                                 BQS1),
-            {true, Confirm, State#q{backing_queue_state = BQS2}};
-        {Duplicate, BQS1} ->
-            Delivered = case Duplicate of
-                            published -> true;
-                            discarded -> false
-                        end,
-            {Delivered, Confirm, State#q{backing_queue_state = BQS1}}
     end.
 
 deliver_or_enqueue(Delivery = #delivery{message = Message,
@@ -652,7 +616,7 @@ handle_ch_down(DownPid, State = #q{exclusive_consumer = Holder}) ->
     case lookup_ch(DownPid) of
         not_found ->
             {ok, State};
-        C = #cr{ch_pid = ChPid, txn = Txn, acktags = ChAckTags} ->
+        C = #cr{ch_pid = ChPid, acktags = ChAckTags} ->
             ok = erase_ch_record(C),
             State1 = State#q{
                        exclusive_consumer = case Holder of
@@ -665,13 +629,8 @@ handle_ch_down(DownPid, State = #q{exclusive_consumer = Holder}) ->
                                              ChPid, State#q.blocked_consumers)},
             case should_auto_delete(State1) of
                 true  -> {stop, State1};
-                false -> State2 = case Txn of
-                                      none -> State1;
-                                      _    -> rollback_transaction(Txn, C,
-                                                                   State1)
-                                  end,
-                         {ok, requeue_and_run(sets:to_list(ChAckTags),
-                                              ensure_expiry_timer(State2))}
+                false -> {ok, requeue_and_run(sets:to_list(ChAckTags),
+                                              ensure_expiry_timer(State1))}
             end
     end.
 
@@ -705,25 +664,6 @@ run_backing_queue(Mod, Fun, State = #q{backing_queue = BQ,
                                        backing_queue_state = BQS}) ->
     run_message_queue(State#q{backing_queue_state = BQ:invoke(Mod, Fun, BQS)}).
 
-commit_transaction(Txn, From, C = #cr{acktags = ChAckTags},
-                   State = #q{backing_queue       = BQ,
-                              backing_queue_state = BQS,
-                              ttl                 = TTL}) ->
-    {AckTags, BQS1} = BQ:tx_commit(
-                        Txn, fun () -> gen_server2:reply(From, ok) end,
-                        reset_msg_expiry_fun(TTL), BQS),
-    ChAckTags1 = subtract_acks(ChAckTags, AckTags),
-    maybe_store_ch_record(C#cr{acktags = ChAckTags1, txn = none}),
-    State#q{backing_queue_state = BQS1}.
-
-rollback_transaction(Txn, C, State = #q{backing_queue = BQ,
-                                        backing_queue_state = BQS}) ->
-    {_AckTags, BQS1} = BQ:tx_rollback(Txn, BQS),
-    %% Iff we removed acktags from the channel record on ack+txn then
-    %% we would add them back in here.
-    maybe_store_ch_record(C#cr{txn = none}),
-    State#q{backing_queue_state = BQS1}.
-
 subtract_acks(A, B) when is_list(B) ->
     lists:foldl(fun sets:del_element/2, A, B).
 
@@ -933,13 +873,6 @@ handle_call({deliver, Delivery}, From, State) ->
     gen_server2:reply(From, true),
     noreply(deliver_or_enqueue(Delivery, State));
 
-handle_call({commit, Txn, ChPid}, From, State) ->
-    case lookup_ch(ChPid) of
-        not_found -> reply(ok, State);
-        C         -> noreply(run_message_queue(
-                               commit_transaction(Txn, From, C, State)))
-    end;
-
 handle_call({notify_down, ChPid}, _From, State) ->
     %% we want to do this synchronously, so that auto_deleted queues
     %% are no longer visible by the time we send a response to the
@@ -1095,24 +1028,16 @@ handle_cast({deliver, Delivery}, State) ->
     %% Asynchronous, non-"mandatory", non-"immediate" deliver mode.
     noreply(deliver_or_enqueue(Delivery, State));
 
-handle_cast({ack, Txn, AckTags, ChPid},
+handle_cast({ack, none, AckTags, ChPid},
             State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
     case lookup_ch(ChPid) of
         not_found ->
             noreply(State);
         C = #cr{acktags = ChAckTags} ->
-            {C1, State1} =
-                case Txn of
-                    none -> ChAckTags1 = subtract_acks(ChAckTags, AckTags),
-                            NewC = C#cr{acktags = ChAckTags1},
-                            {_Guids, BQS1} = BQ:ack(AckTags, BQS),
-                            {NewC, State#q{backing_queue_state = BQS1}};
-                    _    -> BQS1 = BQ:tx_ack(Txn, AckTags, BQS),
-                            {C#cr{txn = Txn},
-                             State#q{backing_queue_state = BQS1}}
-                end,
-            maybe_store_ch_record(C1),
-            noreply(State1)
+            maybe_store_ch_record(C#cr{acktags = subtract_acks(
+                                                   ChAckTags, AckTags)}),
+            {_Guids, BQS1} = BQ:ack(AckTags, BQS),
+            noreply(State#q{backing_queue_state = BQS1})
     end;
 
 handle_cast({reject, AckTags, Requeue, ChPid},
@@ -1131,12 +1056,6 @@ handle_cast({reject, AckTags, Requeue, ChPid},
                     end)
     end;
 
-handle_cast({rollback, Txn, ChPid}, State) ->
-    noreply(case lookup_ch(ChPid) of
-                not_found -> State;
-                C         -> rollback_transaction(Txn, C, State)
-            end);
-
 handle_cast(delete_immediately, State) ->
     {stop, normal, State};
 
diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl
index 217ad3eb..77278416 100644
--- a/src/rabbit_backing_queue.erl
+++ b/src/rabbit_backing_queue.erl
@@ -44,9 +44,7 @@ behaviour_info(callbacks) ->
      %%    makes it useful for passing messages back into the backing
      %%    queue, especially as the backing queue does not have
      %%    control of its own mailbox.
-     %% 4. a synchronous callback. Same as the asynchronous callback
-     %%    but waits for completion and returns 'error' on error.
-     {init, 4},
+     {init, 3},
 
      %% Called on queue shutdown when queue isn't being deleted.
      {terminate, 2},
@@ -107,21 +105,6 @@ behaviour_info(callbacks) ->
      %% about. Must return 1 msg_id per Ack, in the same order as Acks.
      {ack, 2},
 
-     %% A publish, but in the context of a transaction.
-     {tx_publish, 5},
-
-     %% Acks, but in the context of a transaction.
-     {tx_ack, 3},
-
-     %% Undo anything which has been done in the context of the
-     %% specified transaction.
-     {tx_rollback, 2},
-
-     %% Commit a transaction. The Fun passed in must be called once
-     %% the messages have really been commited. This CPS permits the
-     %% possibility of commit coalescing.
-     {tx_commit, 4},
-
      %% Reinsert messages into the queue which have already been
      %% delivered and were pending acknowledgement.
      {requeue, 3},
@@ -175,7 +158,7 @@ behaviour_info(callbacks) ->
      %% the BQ to signal that it's already seen this message (and in
      %% what capacity - i.e. was it published previously or discarded
      %% previously) and thus the message should be dropped.
-     {is_duplicate, 3},
+     {is_duplicate, 2},
 
      %% Called to inform the BQ about messages which have reached the
      %% queue, but are not going to be further passed to BQ for some
diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl
index 463b8cfb..082730e0 100644
--- a/src/rabbit_mirror_queue_master.erl
+++ b/src/rabbit_mirror_queue_master.erl
@@ -16,13 +16,12 @@
 
 -module(rabbit_mirror_queue_master).
 
--export([init/4, terminate/2, delete_and_terminate/2,
+-export([init/3, terminate/2, delete_and_terminate/2,
          purge/1, publish/4, publish_delivered/5, fetch/2, ack/2,
-         tx_publish/5, tx_ack/3, tx_rollback/2, tx_commit/4,
          requeue/3, len/1, is_empty/1, drain_confirmed/1, dropwhile/2,
          set_ram_duration_target/2, ram_duration/1,
          needs_timeout/1, timeout/1, handle_pre_hibernate/1,
-         status/1, invoke/3, is_duplicate/3, discard/3]).
+         status/1, invoke/3, is_duplicate/2, discard/3]).
 
 -export([start/1, stop/0]).
 
@@ -72,7 +71,7 @@ sender_death_fun() ->
     end.
 
 init(#amqqueue { name = QName, mirror_nodes = MNodes } = Q, Recover,
-     AsyncCallback, SyncCallback) ->
+     AsyncCallback) ->
     {ok, CPid} = rabbit_mirror_queue_coordinator:start_link(
                    Q, undefined, sender_death_fun()),
     GM = rabbit_mirror_queue_coordinator:get_gm(CPid),
@@ -84,7 +83,7 @@ init(#amqqueue { name = QName, mirror_nodes = MNodes } = Q, Recover,
          end) -- [node()],
     [rabbit_mirror_queue_misc:add_mirror(QName, Node) || Node <- MNodes1],
     {ok, BQ} = application:get_env(backing_queue_module),
-    BQS = BQ:init(Q, Recover, AsyncCallback, SyncCallback),
+    BQS = BQ:init(Q, Recover, AsyncCallback),
     #state { gm                  = GM,
              coordinator         = CPid,
              backing_queue       = BQ,
@@ -243,21 +242,6 @@ ack(AckTags, State = #state { gm                  = GM,
     {MsgIds, State #state { backing_queue_state = BQS1,
                             ack_msg_id          = AM1 }}.
 
-tx_publish(_Txn, _Msg, _MsgProps, _ChPid, State) ->
-    %% We don't support txns in mirror queues
-    State.
-
-tx_ack(_Txn, _AckTags, State) ->
-    %% We don't support txns in mirror queues
-    State.
-
-tx_rollback(_Txn, State) ->
-    {[], State}.
-
-tx_commit(_Txn, PostCommitFun, _MsgPropsFun, State) ->
-    PostCommitFun(), %% Probably must run it to avoid deadlocks
-    {[], State}.
-
 requeue(AckTags, MsgPropsFun, State = #state { gm                  = GM,
                                                backing_queue       = BQ,
                                                backing_queue_state = BQS }) ->
@@ -299,7 +283,7 @@ invoke(Mod, Fun, State = #state { backing_queue       = BQ,
                                   backing_queue_state = BQS }) ->
     State #state { backing_queue_state = BQ:invoke(Mod, Fun, BQS) }.
 
-is_duplicate(none, Message = #basic_message { id = MsgId },
+is_duplicate(Message = #basic_message { id = MsgId },
              State = #state { seen_status         = SS,
                               backing_queue       = BQ,
                               backing_queue_state = BQS,
@@ -341,11 +325,7 @@ is_duplicate(none, Message = #basic_message { id = MsgId },
             %% Don't erase from SS here because discard/2 is about to
             %% be called and we need to be able to detect this case
             {discarded, State}
-    end;
-is_duplicate(_Txn, _Msg, State) ->
-    %% In a transaction. We don't support txns in mirror queues. But
-    %% it's probably not a duplicate...
-    {false, State}.
+    end.
 
 discard(Msg = #basic_message { id = MsgId }, ChPid,
         State = #state { gm                  = GM,
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index f5492cdc..6e44c7a0 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -1669,8 +1669,9 @@ test_backing_queue() ->
             passed = test_queue_index(),
             passed = test_queue_index_props(),
             passed = test_variable_queue(),
-            passed = test_variable_queue_delete_msg_store_files_callback(),
-            passed = test_queue_recover(),
+            %% FIXME: replace the use of tx in these with confirms
+            %% passed = test_variable_queue_delete_msg_store_files_callback(),
+            %% passed = test_queue_recover(),
             application:set_env(rabbit, queue_index_max_journal_entries,
                                 MaxJournal, infinity),
             passed;
@@ -2084,7 +2085,7 @@ test_queue_index() ->
 
 variable_queue_init(Q, Recover) ->
     rabbit_variable_queue:init(
-      Q, Recover, fun nop/2, fun nop/2, fun nop/2, fun nop/1).
+      Q, Recover, fun nop/2, fun nop/2, fun nop/1).
 
 variable_queue_publish(IsPersistent, Count, VQ) ->
     lists:foldl(
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index a167cca0..630be00b 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -16,19 +16,18 @@
 
 -module(rabbit_variable_queue).
 
--export([init/4, terminate/2, delete_and_terminate/2,
+-export([init/3, terminate/2, delete_and_terminate/2,
          purge/1, publish/4, publish_delivered/5, drain_confirmed/1,
-         fetch/2, ack/2, tx_publish/5, tx_ack/3, tx_rollback/2, tx_commit/4,
-         requeue/3, len/1, is_empty/1, dropwhile/2,
+         fetch/2, ack/2, requeue/3, len/1, is_empty/1, dropwhile/2,
          set_ram_duration_target/2, ram_duration/1,
          needs_timeout/1, timeout/1, handle_pre_hibernate/1,
-         status/1, invoke/3, is_duplicate/3, discard/3,
+         status/1, invoke/3, is_duplicate/2, discard/3,
          multiple_routing_keys/0]).
 
 -export([start/1, stop/0]).
 
 %% exported for testing only
--export([start_msg_store/2, stop_msg_store/0, init/6]).
+-export([start_msg_store/2, stop_msg_store/0, init/5]).
 
 %%----------------------------------------------------------------------------
 %% Definitions:
@@ -238,12 +237,10 @@
           ram_ack_index,
           index_state,
           msg_store_clients,
-          on_sync,
           durable,
           transient_threshold,
 
           async_callback,
-          sync_callback,
 
           len,
           persistent_count,
@@ -284,10 +281,6 @@
           end_seq_id    %% end_seq_id is exclusive
         }).
 
--record(tx, { pending_messages, pending_acks }).
-
--record(sync, { acks_persistent, acks_all, pubs, funs }).
-
 %% When we discover, on publish, that we should write some indices to
 %% disk for some betas, the IO_BATCH_SIZE sets the number of betas
 %% that we must be due to write indices for before we do any work at
@@ -320,12 +313,6 @@
                           count        :: non_neg_integer(),
                           end_seq_id   :: non_neg_integer() }).
 
--type(sync() :: #sync { acks_persistent :: [[seq_id()]],
-                        acks_all        :: [[seq_id()]],
-                        pubs            :: [{message_properties_transformer(),
-                                             [rabbit_types:basic_message()]}],
-                        funs            :: [fun (() -> any())] }).
-
 -type(state() :: #vqstate {
              q1                    :: queue(),
              q2                    :: bpqueue:bpqueue(),
@@ -338,12 +325,10 @@
              index_state           :: any(),
              msg_store_clients     :: 'undefined' | {{any(), binary()},
                                                     {any(), binary()}},
-             on_sync               :: sync(),
              durable               :: boolean(),
              transient_threshold   :: non_neg_integer(),
 
              async_callback        :: async_callback(),
-             sync_callback         :: sync_callback(),
 
              len                   :: non_neg_integer(),
              persistent_count      :: non_neg_integer(),
@@ -376,11 +361,6 @@
                                          count        = 0,
                                          end_seq_id   = Z }).
 
--define(BLANK_SYNC, #sync { acks_persistent = [],
-                            acks_all        = [],
-                            pubs            = [],
-                            funs            = [] }).
-
 %%----------------------------------------------------------------------------
 %% Public API
 %%----------------------------------------------------------------------------
@@ -409,17 +389,17 @@ stop_msg_store() ->
     ok = rabbit_sup:stop_child(?PERSISTENT_MSG_STORE),
     ok = rabbit_sup:stop_child(?TRANSIENT_MSG_STORE).
 
-init(Queue, Recover, AsyncCallback, SyncCallback) ->
-    init(Queue, Recover, AsyncCallback, SyncCallback,
+init(Queue, Recover, AsyncCallback) ->
+    init(Queue, Recover, AsyncCallback,
          fun (MsgIds, ActionTaken) ->
                  msgs_written_to_disk(AsyncCallback, MsgIds, ActionTaken)
          end,
          fun (MsgIds) -> msg_indices_written_to_disk(AsyncCallback, MsgIds) end).
 
 init(#amqqueue { name = QueueName, durable = IsDurable }, false,
-     AsyncCallback, SyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) ->
+     AsyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) ->
     IndexState = rabbit_queue_index:init(QueueName, MsgIdxOnDiskFun),
-    init(IsDurable, IndexState, 0, [], AsyncCallback, SyncCallback,
+    init(IsDurable, IndexState, 0, [], AsyncCallback,
          case IsDurable of
              true  -> msg_store_client_init(?PERSISTENT_MSG_STORE,
                                             MsgOnDiskFun, AsyncCallback);
@@ -428,7 +408,7 @@ init(#amqqueue { name = QueueName, durable = IsDurable }, false,
          msg_store_client_init(?TRANSIENT_MSG_STORE, undefined, AsyncCallback));
 
 init(#amqqueue { name = QueueName, durable = true }, true,
-     AsyncCallback, SyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) ->
+     AsyncCallback, MsgOnDiskFun, MsgIdxOnDiskFun) ->
     Terms = rabbit_queue_index:shutdown_terms(QueueName),
     {PRef, TRef, Terms1} =
         case [persistent_ref, transient_ref] -- proplists:get_keys(Terms) of
@@ -449,14 +429,14 @@ init(#amqqueue { name = QueueName, durable = true }, true,
                   rabbit_msg_store:contains(MsgId, PersistentClient)
           end,
           MsgIdxOnDiskFun),
-    init(true, IndexState, DeltaCount, Terms1, AsyncCallback, SyncCallback,
+    init(true, IndexState, DeltaCount, Terms1, AsyncCallback,
          PersistentClient, TransientClient).
 
 terminate(_Reason, State) ->
     State1 = #vqstate { persistent_count  = PCount,
                         index_state       = IndexState,
                         msg_store_clients = {MSCStateP, MSCStateT} } =
-        remove_pending_ack(true, tx_commit_index(State)),
+        remove_pending_ack(true, State),
     PRef = case MSCStateP of
                undefined -> undefined;
                _         -> ok = rabbit_msg_store:client_terminate(MSCStateP),
@@ -674,59 +654,6 @@ ack(AckTags, State) ->
                            AckTags, State),
     {MsgIds, a(State1)}.
 
-tx_publish(Txn, Msg = #basic_message { is_persistent = IsPersistent }, MsgProps,
-           _ChPid, State = #vqstate { durable           = IsDurable,
-                                      msg_store_clients = MSCState }) ->
-    Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn),
-    store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProps} | Pubs] }),
-    case IsPersistent andalso IsDurable of
-        true  -> MsgStatus = msg_status(true, undefined, Msg, MsgProps),
-                 #msg_status { msg_on_disk = true } =
-                     maybe_write_msg_to_disk(false, MsgStatus, MSCState);
-        false -> ok
-    end,
-    a(State).
-
-tx_ack(Txn, AckTags, State) ->
-    Tx = #tx { pending_acks = Acks } = lookup_tx(Txn),
-    store_tx(Txn, Tx #tx { pending_acks = [AckTags | Acks] }),
-    State.
-
-tx_rollback(Txn, State = #vqstate { durable           = IsDurable,
-                                    msg_store_clients = MSCState }) ->
-    #tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn),
-    erase_tx(Txn),
-    ok = case IsDurable of
-             true  -> msg_store_remove(MSCState, true,
-                                       persistent_msg_ids(Pubs));
-             false -> ok
-         end,
-    {lists:append(AckTags), a(State)}.
-
-tx_commit(Txn, Fun, MsgPropsFun,
-          State = #vqstate { durable           = IsDurable,
-                             async_callback    = AsyncCallback,
-                             sync_callback     = SyncCallback,
-                             msg_store_clients = MSCState }) ->
-    #tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn),
-    erase_tx(Txn),
-    AckTags1 = lists:append(AckTags),
-    PersistentMsgIds = persistent_msg_ids(Pubs),
-    HasPersistentPubs = PersistentMsgIds =/= [],
-    {AckTags1,
-     a(case IsDurable andalso HasPersistentPubs of
-           true  -> MsgStoreCallback =
-                        fun () -> msg_store_callback(
-                                    PersistentMsgIds, Pubs, AckTags1, Fun,
-                                    MsgPropsFun, AsyncCallback, SyncCallback)
-                        end,
-                    ok = msg_store_sync(MSCState, true, PersistentMsgIds,
-                                        fun () -> spawn(MsgStoreCallback) end),
-                    State;
-           false -> tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags1,
-                                             Fun, MsgPropsFun, State)
-       end)}.
-
 requeue(AckTags, MsgPropsFun, State) ->
     MsgPropsFun1 = fun (MsgProps) ->
                            (MsgPropsFun(MsgProps)) #message_properties {
@@ -832,23 +759,22 @@ ram_duration(State = #vqstate {
                  ram_msg_count_prev = RamMsgCount,
                  ram_ack_count_prev = RamAckCount }}.
 
-needs_timeout(State = #vqstate { on_sync = OnSync }) ->
-    case {OnSync, needs_index_sync(State)} of
-        {?BLANK_SYNC, false} ->
-            case 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) of
-                {true,  _State} -> idle;
-                {false, _State} -> false
-            end;
-        _ ->
-            timed
+needs_timeout(State) ->
+    case needs_index_sync(State) of
+        false -> case 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) of
+                     {true,  _State} -> idle;
+                     {false, _State} -> false
+                 end;
+        true  -> timed
     end.
 
 timeout(State) ->
-    a(reduce_memory_use(confirm_commit_index(tx_commit_index(State)))).
+    a(reduce_memory_use(confirm_commit_index(State))).
 
 handle_pre_hibernate(State = #vqstate { index_state = IndexState }) ->
     State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }.
@@ -858,7 +784,6 @@ status(#vqstate {
           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,
@@ -875,7 +800,6 @@ status(#vqstate {
       {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)},
@@ -887,10 +811,9 @@ status(#vqstate {
       {avg_ack_ingress_rate, AvgAckIngressRate},
       {avg_ack_egress_rate , AvgAckEgressRate} ].
 
-invoke(?MODULE, Fun, State) ->
-    Fun(?MODULE, State).
+invoke(?MODULE, Fun, State) -> Fun(?MODULE, State).
 
-is_duplicate(_Txn, _Msg, State) -> {false, State}.
+is_duplicate(_Msg, State) -> {false, State}.
 
 discard(_Msg, _ChPid, State) -> State.
 
@@ -986,11 +909,6 @@ msg_store_remove(MSCState, IsPersistent, MsgIds) ->
       MSCState, IsPersistent,
       fun (MCSState1) -> rabbit_msg_store:remove(MsgIds, MCSState1) end).
 
-msg_store_sync(MSCState, IsPersistent, MsgIds, Fun) ->
-    with_immutable_msg_store_state(
-      MSCState, IsPersistent,
-      fun (MSCState1) -> rabbit_msg_store:sync(MsgIds, Fun, MSCState1) end).
-
 msg_store_close_fds(MSCState, IsPersistent) ->
     with_msg_store_state(
       MSCState, IsPersistent,
@@ -1007,20 +925,6 @@ maybe_write_delivered(false, _SeqId, IndexState) ->
 maybe_write_delivered(true, SeqId, IndexState) ->
     rabbit_queue_index:deliver([SeqId], IndexState).
 
-lookup_tx(Txn) -> case get({txn, Txn}) of
-                      undefined -> #tx { pending_messages = [],
-                                         pending_acks     = [] };
-                      V         -> V
-                  end.
-
-store_tx(Txn, Tx) -> put({txn, Txn}, Tx).
-
-erase_tx(Txn) -> erase({txn, Txn}).
-
-persistent_msg_ids(Pubs) ->
-    [MsgId || {#basic_message { id            = MsgId,
-                                is_persistent = true }, _MsgProps} <- Pubs].
-
 betas_from_index_entries(List, TransientThreshold, IndexState) ->
     {Filtered, Delivers, Acks} =
         lists:foldr(
@@ -1084,8 +988,8 @@ update_rate(Now, Then, Count, {OThen, OCount}) ->
 %% Internal major helpers for Public API
 %%----------------------------------------------------------------------------
 
-init(IsDurable, IndexState, DeltaCount, Terms,
-     AsyncCallback, SyncCallback, PersistentClient, TransientClient) ->
+init(IsDurable, IndexState, DeltaCount, Terms, AsyncCallback,
+     PersistentClient, TransientClient) ->
     {LowSeqId, NextSeqId, IndexState1} = rabbit_queue_index:bounds(IndexState),
 
     DeltaCount1 = proplists:get_value(persistent_count, Terms, DeltaCount),
@@ -1107,12 +1011,10 @@ init(IsDurable, IndexState, DeltaCount, Terms,
       ram_ack_index       = gb_trees:empty(),
       index_state         = IndexState1,
       msg_store_clients   = {PersistentClient, TransientClient},
-      on_sync             = ?BLANK_SYNC,
       durable             = IsDurable,
       transient_threshold = NextSeqId,
 
       async_callback      = AsyncCallback,
-      sync_callback       = SyncCallback,
 
       len                 = DeltaCount1,
       persistent_count    = DeltaCount1,
@@ -1141,88 +1043,6 @@ blank_rate(Timestamp, IngressLength) ->
              avg_ingress = 0.0,
              timestamp   = Timestamp }.
 
-msg_store_callback(PersistentMsgIds, Pubs, AckTags, Fun, MsgPropsFun,
-                   AsyncCallback, SyncCallback) ->
-    case SyncCallback(?MODULE,
-                      fun (?MODULE, StateN) ->
-                              tx_commit_post_msg_store(true, Pubs, AckTags,
-                                                       Fun, MsgPropsFun, StateN)
-                      end) of
-        ok    -> ok;
-        error -> remove_persistent_messages(PersistentMsgIds, AsyncCallback)
-    end.
-
-remove_persistent_messages(MsgIds, AsyncCallback) ->
-    PersistentClient = msg_store_client_init(?PERSISTENT_MSG_STORE,
-                                             undefined, AsyncCallback),
-    ok = rabbit_msg_store:remove(MsgIds, PersistentClient),
-    rabbit_msg_store:client_delete_and_terminate(PersistentClient).
-
-tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun, MsgPropsFun,
-                         State = #vqstate {
-                           on_sync     = OnSync = #sync {
-                                           acks_persistent = SPAcks,
-                                           acks_all        = SAcks,
-                                           pubs            = SPubs,
-                                           funs            = SFuns },
-                           pending_ack = PA,
-                           durable     = IsDurable }) ->
-    PersistentAcks =
-        case IsDurable of
-            true  -> [AckTag || AckTag <- AckTags,
-                                case dict:fetch(AckTag, PA) of
-                                    #msg_status {} ->
-                                        false;
-                                    {IsPersistent, _MsgId, _MsgProps} ->
-                                        IsPersistent
-                                end];
-            false -> []
-        end,
-    case IsDurable andalso (HasPersistentPubs orelse PersistentAcks =/= []) of
-        true  -> State #vqstate {
-                   on_sync = #sync {
-                     acks_persistent = [PersistentAcks | SPAcks],
-                     acks_all        = [AckTags | SAcks],
-                     pubs            = [{MsgPropsFun, Pubs} | SPubs],
-                     funs            = [Fun | SFuns] }};
-        false -> State1 = tx_commit_index(
-                            State #vqstate {
-                              on_sync = #sync {
-                                acks_persistent = [],
-                                acks_all        = [AckTags],
-                                pubs            = [{MsgPropsFun, Pubs}],
-                                funs            = [Fun] } }),
-                 State1 #vqstate { on_sync = OnSync }
-    end.
-
-tx_commit_index(State = #vqstate { on_sync = ?BLANK_SYNC }) ->
-    State;
-tx_commit_index(State = #vqstate { on_sync = #sync {
-                                     acks_persistent = SPAcks,
-                                     acks_all        = SAcks,
-                                     pubs            = SPubs,
-                                     funs            = SFuns },
-                                   durable = IsDurable }) ->
-    PAcks = lists:append(SPAcks),
-    Acks  = lists:append(SAcks),
-    Pubs  = [{Msg, Fun(MsgProps)} || {Fun, PubsN}    <- lists:reverse(SPubs),
-                                     {Msg, MsgProps} <- lists:reverse(PubsN)],
-    {_MsgIds, State1} = ack(Acks, State),
-    {SeqIds, State2 = #vqstate { index_state = IndexState }} =
-        lists:foldl(
-          fun ({Msg = #basic_message { is_persistent = IsPersistent },
-                MsgProps},
-               {SeqIdsAcc, State3}) ->
-                  IsPersistent1 = IsDurable andalso IsPersistent,
-                  {SeqId, State4} =
-                      publish(Msg, MsgProps, false, IsPersistent1, State3),
-                  {cons_if(IsPersistent1, SeqId, SeqIdsAcc), State4}
-          end, {PAcks, State1}, Pubs),
-    IndexState1 = rabbit_queue_index:sync(SeqIds, IndexState),
-    [ Fun() || Fun <- lists:reverse(SFuns) ],
-    reduce_memory_use(
-      State2 #vqstate { index_state = IndexState1, on_sync = ?BLANK_SYNC }).
-
 purge_betas_and_deltas(LensByStore,
                        State = #vqstate { q3                = Q3,
                                           index_state       = IndexState,