Repro hacks

Author: the-mikedavis

deps/rabbit/Makefile

@@ -129,8 +129,8 @@ endef
 LOCAL_DEPS = sasl os_mon inets compiler public_key crypto ssl syntax_tools xmerl
 
 BUILD_DEPS = rabbitmq_cli
-DEPS = ranch cowlib rabbit_common amqp10_common rabbitmq_prelaunch ra sysmon_handler stdout_formatter recon redbug observer_cli osiris syslog systemd seshat horus khepri khepri_mnesia_migration cuttlefish gen_batch_server
-TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers meck proper amqp_client rabbitmq_amqp_client rabbitmq_amqp1_0
+DEPS = ranch cowlib rabbit_common amqp10_common rabbitmq_prelaunch ra sysmon_handler stdout_formatter recon redbug observer_cli osiris syslog systemd seshat horus khepri khepri_mnesia_migration cuttlefish gen_batch_server amqp_client
+TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers meck proper rabbitmq_amqp_client rabbitmq_amqp1_0
 
 # We pin a version of Horus even if we don't use it directly (it is a
 # dependency of Khepri). But currently, we can't update Khepri while still

deps/rabbit/src/rabbit_fifo.erl

@@ -726,9 +726,12 @@ handle_down(Meta, Pid, #?STATE{consumers = Cons0,
     DownConsumers = maps:keys(maps:filter(fun(_CKey, ?CONSUMER_PID(P)) ->
                                                   P =:= Pid
                                           end, Cons0)),
+    %% Shuffle the map keys to simulate the keys having different orderings
+    %% among cluster members.
+    DownConsumers1 = [X || {_, X} <- lists:sort([{rand:uniform(), N} || N <- DownConsumers])],
     lists:foldl(fun(ConsumerKey, {S, E}) ->
                         cancel_consumer(Meta, ConsumerKey, S, E, down)
-                end, {State2, Effects1}, DownConsumers).
+                end, {State2, Effects1}, DownConsumers1).
 
 consumer_active_flag_update_function(
   #?STATE{cfg = #cfg{consumer_strategy = competing}}) ->

deps/rabbit/src/rabbit_repro.erl

@@ -0,0 +1,78 @@
+-module(rabbit_repro).
+
+-export([run/0]).
+
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+-define(Q, <<"qq">>).
+-define(N_MESSAGES, 10).
+
+run() ->
+    {_, BasicGetRunnerRef} = spawn_monitor(fun() ->
+        {ok, Conn} = connection(),
+        link(Conn),
+        {ok, Ch} = amqp_connection:open_channel(Conn),
+        link(Ch),
+
+        %% Get 10 messages and then exit without checking them out so that
+        %% they are returned.
+        basic_get(Ch, ?N_MESSAGES),
+        ok
+    end),
+
+    receive
+        {'DOWN', BasicGetRunnerRef, process, _, normal} ->
+            ok;
+        {'DOWN', BasicGetRunnerRef, process, _, _} ->
+            exit(basic_get_runner_fail)
+    end,
+
+    %% Probably not necessary but let's let everything settle. Wait for 1/2
+    %% second.
+    timer:sleep(500),
+
+    {_, BasicRejectRunnerRef} = spawn_monitor(fun() ->
+        {ok, Conn} = connection(),
+        link(Conn),
+        {ok, Ch} = amqp_connection:open_channel(Conn),
+        link(Ch),
+
+        %% Get and ack 9 messages.
+        {some, AckDeliveryTag} = basic_get(Ch, ?N_MESSAGES - 1),
+        amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = AckDeliveryTag,
+                                           multiple = true}),
+
+        %% Then get and reject the last returned message.
+        {some, RejectDeliveryTag} = basic_get(Ch, 1),
+        amqp_channel:cast(Ch, #'basic.reject'{delivery_tag = RejectDeliveryTag,
+                                              requeue = true}),
+        ok
+    end),
+
+    receive
+        {'DOWN', BasicRejectRunnerRef, process, _, normal} ->
+            ok;
+        {'DOWN', BasicRejectRunnerRef, process, _, _} ->
+            exit(basic_reject_runner_fail)
+    end,
+
+    ok.
+
+connection() ->
+    amqp_connection:start(#amqp_params_direct{virtual_host = <<"/">>,
+                                              username = <<"guest">>,
+                                              password = <<"guest">>}).
+
+basic_get(Ch, N) ->
+    basic_get(Ch, N, none).
+
+basic_get(_Ch, 0, Tag) ->
+    Tag;
+basic_get(Ch, N, Tag0) ->
+    Tag = case amqp_channel:call(Ch, #'basic.get'{queue = ?Q}) of
+        {#'basic.get_ok'{delivery_tag = DeliveryTag}, _} ->
+            {some, DeliveryTag};
+        #'basic.get_empty'{} ->
+            Tag0
+    end,
+    basic_get(Ch, N - 1, Tag).