QQ: fix bug when subscribing using an already existing consumer tag

When subscribing using a consumer tag that is already in the quorum
queues state (but perhaps with a cancelled status) and that has
pending messages the next_msg_id which is used to initialise the
queue type consumer state did not take the in flight message ids into
account. This resulted in some messages occasionally not being delivered
to the clint and thus would appear stuck as awaiting acknowledgement
for the consumer.

When a new checkout operation detects there are in-flight messages
we set the last_msg_id to `undefined` and just accept the next message
that arrives, irrespective of their message id. This isn't 100% fool proof
as there may be cases where messages are lost between queue and channel
where we'd miss to trigger the fallback query for missing messages.

It is however much better than what we have atm.

NB: really the ideal solution would be to make checkout operations
async so that any inflight messages are delivered before the checkout
result. That is a much bigger change for another day.

(cherry picked from commit 49108a6)

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -408,6 +408,7 @@ apply(Meta, #checkout{spec = Spec, meta = ConsumerMeta,
               credit = Credit,
               delivery_count = DeliveryCount,
               next_msg_id = NextMsgId} = Consumer,
+
     %% reply with a consumer summary
     Reply = {ok, #{next_msg_id => NextMsgId,
                    credit => Credit,

deps/rabbit/src/rabbit_fifo_client.erl

@@ -46,7 +46,7 @@
                   rabbit_queue_type:action().
 -type actions() :: [action()].
 
--record(consumer, {last_msg_id :: seq() | -1,
+-record(consumer, {last_msg_id :: seq() | -1 | undefined,
                    ack = false :: boolean(),
                    delivery_count = 0 :: non_neg_integer()}).
 
@@ -359,8 +359,20 @@ checkout(ConsumerTag, NumUnsettled, CreditMode, Meta,
                                 %% this is the pre 3.11.1 / 3.10.9
                                 %% reply format
                                 -1;
-                            {ok, #{next_msg_id := NextMsgId}} ->
-                                NextMsgId - 1
+                            {ok, #{num_checked_out := NumChecked,
+                                   next_msg_id := NextMsgId}} ->
+                                case NumChecked > 0 of
+                                    true ->
+                                        %% we cannot know if the pending messages
+                                        %% have been delivered to the client or they
+                                        %% are on their way to the current process.
+                                        %% We set `undefined' to signal this uncertainty
+                                        %% and will just accept the next arriving message
+                                        %% irrespective of message id
+                                        undefined;
+                                    false ->
+                                        NextMsgId - 1
+                                end
                         end,
             SDels = maps:update_with(
                       ConsumerTag, fun (C) -> C#consumer{ack = Ack} end,
@@ -713,7 +725,11 @@ handle_delivery(QName, Leader, {delivery, Tag, [{FstId, _} | _] = IdMsgs},
     %% TODO: remove potential default allocation
     case Consumer of
         #consumer{last_msg_id = Prev} = C
-          when FstId =:= Prev+1 ->
+          when Prev =:= undefined orelse FstId =:= Prev+1 ->
+            %% Prev =:= undefined is a special case where a checkout was done
+            %% for a previously cancelled consumer that still had pending messages
+            %% In this case we can't reliably know what the next expected message
+            %% id should be so have to accept whatever message comes next
             maybe_auto_ack(Ack, Del,
                            State0#state{consumer_deliveries =
                                         update_consumer(Tag, LastId,

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -1726,7 +1726,7 @@ confirm_availability_on_leader_change(Config) ->
 
 flush(T) ->
     receive X ->
-                ct:pal("flushed ~w", [X]),
+                ct:pal("flushed ~p", [X]),
                 flush(T)
     after T ->
               ok
@@ -3057,12 +3057,13 @@ cancel_and_consume_with_same_tag(Config) ->
     DeclareRslt0 = declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
     ?assertMatch(ExpectedDeclareRslt0, DeclareRslt0),
 
-    ok = publish(Ch, QQ),
+    ok = publish(Ch, QQ, <<"msg1">>),
 
     ok = subscribe(Ch, QQ, false),
 
     DeliveryTag = receive
-                      {#'basic.deliver'{delivery_tag = D}, _} ->
+                      {#'basic.deliver'{delivery_tag = D},
+                       #amqp_msg{payload = <<"msg1">>}} ->
                           D
                   after 5000 ->
                             flush(100),
@@ -3073,10 +3074,11 @@ cancel_and_consume_with_same_tag(Config) ->
 
     ok = subscribe(Ch, QQ, false),
 
-    ok = publish(Ch, QQ),
+    ok = publish(Ch, QQ, <<"msg2">>),
 
     receive
-        {#'basic.deliver'{delivery_tag = _}, _} ->
+        {#'basic.deliver'{delivery_tag = _},
+         #amqp_msg{payload = <<"msg2">>}} ->
             ok
     after 5000 ->
               flush(100),

deps/rabbit/test/rabbit_fifo_SUITE.erl

@@ -1990,6 +1990,26 @@ chunk_disk_msgs_test(_Config) ->
                  rabbit_fifo:chunk_disk_msgs(TwoBigMsgs, 0, [[]])),
     ok.
 
+checkout_metadata_test(Config) ->
+    Cid = {<<"cid">>, self()},
+    {State00, _} = enq(Config, 1, 1, first, test_init(test)),
+    {State0, _} = enq(Config, 2, 2, second, State00),
+    %% NB: the consumer meta data is taken _before_ it runs a checkout
+    %% so in this case num_checked_out will be 0
+    {State1, {ok, #{next_msg_id := 0,
+                    num_checked_out := 0}}, _} =
+        apply(meta(Config, ?LINE),
+              rabbit_fifo:make_checkout(Cid, {auto, 1, simple_prefetch}, #{}),
+              State0),
+    {State2, _, _} = apply(meta(Config, ?LINE),
+                           rabbit_fifo:make_checkout(Cid, cancel, #{}), State1),
+    {_State3, {ok, #{next_msg_id := 1,
+                     num_checked_out := 1}}, _} =
+        apply(meta(Config, ?LINE),
+              rabbit_fifo:make_checkout(Cid, {auto, 1, simple_prefetch}, #{}),
+              State2),
+    ok.
+
 %% Utility
 
 init(Conf) -> rabbit_fifo:init(Conf).