substantially bulked up the tests for this bug. All tests pass.

2 files changed, 94 insertions, 48 deletions

diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
index f9a8f488af..63d6a4815d 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, message/5, delivery/4]).
+-export([publish/1, message/4, message/5, message/6, delivery/4]).
 
 %%----------------------------------------------------------------------------
 
@@ -44,8 +44,10 @@
 -spec(delivery/4 :: (bool(), bool(), maybe(txn()), message()) -> delivery()). 
 -spec(message/4 :: (exchange_name(), routing_key(), binary(), binary()) ->
              message()).
--spec(message/5 :: (exchange_name(), routing_key(), binary(), binary(), guid()) ->
-             message()).
+-spec(message/5 :: (exchange_name(), routing_key(), binary(), binary(),
+                    guid()) -> message()).
+-spec(message/6 :: (exchange_name(), routing_key(), binary(), binary(),
+                    guid(), bool()) -> message()).
 
 -endif.
 
@@ -69,6 +71,9 @@ message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin) ->
     message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, rabbit_guid:guid()).
 
 message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, MsgId) ->
+    message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, MsgId, false).
+
+message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, MsgId, IsPersistent) ->
     {ClassId, _MethodId} = rabbit_framing:method_id('basic.publish'),
     Content = #content{class_id = ClassId,
                        properties = #'P_basic'{content_type = ContentTypeBin},
@@ -78,4 +83,4 @@ message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, MsgId) ->
                    routing_key    = RoutingKeyBin,
                    content        = Content,
                    guid           = MsgId,
-                   is_persistent  = false}.
+                   is_persistent  = IsPersistent}.
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 37c0121cda..b00331c2d6 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -1052,55 +1052,91 @@ rdq_test_mixed_queue_modes() ->
     passed.
 
 rdq_test_mode_conversion_mid_txn() ->
+    Payload = <<0:(8*256)>>,
+    MsgIdsA = lists:seq(0,9),
+    MsgsA = [ rabbit_basic:message(x, <<>>, <<>>, Payload, MsgId,
+                                   (0 == MsgId rem 2))
+            || MsgId <- MsgIdsA ],
+    MsgIdsB = lists:seq(10,20),
+    MsgsB = [ rabbit_basic:message(x, <<>>, <<>>, Payload, MsgId,
+                                   (0 == MsgId rem 2))
+            || MsgId <- MsgIdsB ],
+
     rdq_virgin(),
     rdq_start(),
-    Payload = <<0:(8*256)>>,
-    {ok, MS} = rabbit_mixed_queue:init(q, true, mixed),
-    MsgIds = lists:seq(1,10),
-    {MS2, Msgs} =
-        lists:foldl(
-            fun (N, {MS1, Acc}) ->
-                    Msg = rabbit_basic:message(x, <<>>, <<>>, Payload, N),
-                    {ok, MS1a} = rabbit_mixed_queue:tx_publish(Msg, MS1),
-                    {MS1a, [Msg | Acc]}
-            end, {MS, []}, MsgIds),
-    MsgsOrdered = lists:reverse(Msgs),
-    {ok, MS3} = rabbit_mixed_queue:to_disk_only_mode(MsgsOrdered, MS2),
-    {ok, MS4} = rabbit_mixed_queue:tx_commit(MsgsOrdered, [], MS3),
-    MS6 =
-        lists:foldl(
-          fun (N, MS5) ->
-                  Rem = 10 - N,
-                  {{#basic_message { is_persistent = false },
-                    false, _AckTag, Rem},
-                   MS5a} = rabbit_mixed_queue:deliver(MS5),
-                  MS5a
-          end, MS4, MsgIds),
-    0 = rabbit_mixed_queue:length(MS6),
-    {ok, MS7} = rabbit_mixed_queue:init(q, true, disk),
-    {MS9, Msgs1} =
-        lists:foldl(
-            fun (N, {MS8, Acc}) ->
-                    Msg = rabbit_basic:message(x, <<>>, <<>>, Payload, N),
-                    {ok, MS8a} = rabbit_mixed_queue:tx_publish(Msg, MS8),
-                    {MS8a, [Msg | Acc]}
-            end, {MS7, []}, MsgIds),
-    Msgs1Ordered = lists:reverse(Msgs1),
-    {ok, MS10} = rabbit_mixed_queue:to_mixed_mode(Msgs1Ordered, MS9),
-    {ok, MS11} = rabbit_mixed_queue:tx_commit(Msgs1Ordered, [], MS10),
-    MS13 =
-        lists:foldl(
-          fun (N, MS12) ->
-                  Rem = 10 - N,
-                  {{#basic_message { is_persistent = false },
-                    false, _AckTag, Rem},
-                   MS12a} = rabbit_mixed_queue:deliver(MS12),
-                  MS12a
-          end, MS11, MsgIds),
-    0 = rabbit_mixed_queue:length(MS13),
+    {ok, MS0} = rabbit_mixed_queue:init(q, true, mixed),
+    passed = rdq_tx_publish_mixed_alter_commit_get(
+               MS0, MsgsA, MsgsB, fun rabbit_mixed_queue:to_disk_only_mode/2, commit),
+
+    rdq_stop_virgin_start(),
+    {ok, MS1} = rabbit_mixed_queue:init(q, true, mixed),
+    passed = rdq_tx_publish_mixed_alter_commit_get(
+               MS1, MsgsA, MsgsB, fun rabbit_mixed_queue:to_disk_only_mode/2, cancel),
+
+
+    rdq_stop_virgin_start(),
+    {ok, MS2} = rabbit_mixed_queue:init(q, true, disk),
+    passed = rdq_tx_publish_mixed_alter_commit_get(
+               MS2, MsgsA, MsgsB, fun rabbit_mixed_queue:to_mixed_mode/2, commit),
+
+    rdq_stop_virgin_start(),
+    {ok, MS3} = rabbit_mixed_queue:init(q, true, disk),
+    passed = rdq_tx_publish_mixed_alter_commit_get(
+               MS3, MsgsA, MsgsB, fun rabbit_mixed_queue:to_mixed_mode/2, cancel),
+
     rdq_stop(),
     passed.
 
+rdq_tx_publish_mixed_alter_commit_get(MS0, MsgsA, MsgsB, ChangeFun, CommitOrCancel) ->
+    0 = rabbit_mixed_queue:length(MS0),
+    MS2 = lists:foldl(
+            fun (Msg, MS1) ->
+                    {ok, MS1a} = rabbit_mixed_queue:publish(Msg, MS1),
+                    MS1a
+            end, MS0, MsgsA),
+    Len0 = length(MsgsA),
+    Len0 = rabbit_mixed_queue:length(MS2),
+    MS4 = lists:foldl(
+            fun (Msg, MS3) ->
+                    {ok, MS3a} = rabbit_mixed_queue:tx_publish(Msg, MS3),
+                    MS3a
+            end, MS2, MsgsB),
+    Len0 = rabbit_mixed_queue:length(MS4),
+    {ok, MS5} = ChangeFun(MsgsB, MS4),
+    Len0 = rabbit_mixed_queue:length(MS5),
+    {ok, MS9} =
+        case CommitOrCancel of
+            commit ->
+                {ok, MS6} = rabbit_mixed_queue:tx_commit(MsgsB, [], MS5),
+                Len1 = Len0 + length(MsgsB),
+                Len1 = rabbit_mixed_queue:length(MS6),
+                {AckTags, MS8} =
+                    lists:foldl(
+                      fun (Msg, {Acc, MS7}) ->
+                              Rem = Len1 - (Msg #basic_message.guid) - 1,
+                              {{Msg, false, AckTag, Rem}, MS7a} =
+                                  rabbit_mixed_queue:deliver(MS7),
+                              {[AckTag | Acc], MS7a}
+                      end, {[], MS6}, MsgsA ++ MsgsB),
+                0 = rabbit_mixed_queue:length(MS8),
+                rabbit_mixed_queue:ack(lists:reverse(AckTags), MS8);
+            cancel ->
+                {ok, MS6} = rabbit_mixed_queue:tx_cancel(MsgsB, MS5),
+                Len0 = rabbit_mixed_queue:length(MS6),
+                {AckTags, MS8} =
+                    lists:foldl(
+                      fun (Msg, {Acc, MS7}) ->
+                              Rem = Len0 - (Msg #basic_message.guid) - 1,
+                              {{Msg, false, AckTag, Rem}, MS7a} =
+                                  rabbit_mixed_queue:deliver(MS7),
+                              {[AckTag | Acc], MS7a}
+                      end, {[], MS6}, MsgsA),
+                0 = rabbit_mixed_queue:length(MS8),
+                rabbit_mixed_queue:ack(lists:reverse(AckTags), MS8)
+        end,
+    0 = rabbit_mixed_queue:length(MS9),
+    passed.
+
 rdq_time_commands(Funcs) ->
     lists:foreach(fun (F) -> F() end, Funcs).
 
@@ -1119,3 +1155,8 @@ rdq_start() ->
 rdq_stop() ->
     rabbit_disk_queue:stop(),
     timer:sleep(1000).
+
+rdq_stop_virgin_start() ->
+    rdq_stop(),
+    rdq_virgin(),
+    rdq_start().