[rabbitmq-discuss] Old consumer tags delivered after re-subscribing to a queue

Fri Apr 2 17:54:32 BST 2010

Hi Juan,

On Fri, Apr 02, 2010 at 01:37:11PM -0300, Juan Wajnerman wrote:
> On Apr 2, 2010, at 1:15 PM, Matthew Sackman wrote:
> 
> > On Thu, Apr 01, 2010 at 12:05:25PM -0300, Juan Wajnerman wrote:
> >> After unsubscribing from a queue, following "recover" calls will still redeliver unack'd messages through the channel. Even worse, after resubscribing to the queue, the recovered messages are still sent with the old consumer tag.
> > 
> > The documentation of basic.recover is clear:
> > "This method asks the server to redeliver all unacknowledged messages
> > on a specified channel."
> 
> But what should happen if the subscription was canceled before the messages
> were ack'd? The current implementation is still sending the messages with the same consumer tag.

Ahh! Right, now I understand the problem - sorry, I thought you were
saying that with your second subscription, you were getting the same
ctag, but you're actually saying that the ctag used during the recovery
is the same as the (now cancelled) subscription.

Yes, that indeed is a bug - the comment in the code is rather revealing!

handle_method(#'basic.recover'{requeue = false},
              _, State = #ch{ transaction_id = none,
                              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),

> That's right. The omission was completely intentional to reproduce the issue.
> I'm sorry, maybe I wasn't clear enough about this. The idea is to cancel the
> subscription before the message was ack'd and then resubscribe and perform a recover. 

Actually, I think you'll see this bug if you just do: subscribe,
unsubscribe, recover. The second subscription should be unnecessary.

The problem is that the result of recover is a number of deliveries.
They must have a ctag of some sort, and all the clients will rely on
that. (This is clearly not a ruby client bug - apologies for suggesting
it was.) If the subscription has been cancelled, it's very likely all
clients will get upset not matter what you do - both a new and an
existing ctag will be unrecognised by clients as they will have
forgotten about the ctag on the unsubscribe.

Which suggests that the possible outcomes are:
a) Don't permit a basic.recover if the subscription has been cancelled.
   This gets very tricky - if you have multiple subscriptions on the
   same channel, cancel some of them, and then hit recover, what on
   earth do you do then?
b) On cancel, forget about unacked messages from the subscription being
   cancelled - treat it the same as a channel closure and requeue the
   messages.

The problem with both of these is that they're explicitly against the
wording of the spec.

> > Similarly, the documention for basic.cancel (unsubscribe):
> > "This method cancels a consumer. This does not affect already delivered
> > messages, but it does mean the server will not send any more messages for
> > that consumer."
> 
> I understand this, but I'm looking at the bits on the wire, and after sending a cancel, the
> following recover will send the unack'd messages using the same old consumer tag.

Yes. There's probably an argument for saying that in the absence of a
valid consumer tag, the messages should be sent to the client as a
sequence of basic.get-oks, or at least a basic.deliver with a very
special ctag to indicate to the clients that they shouldn't be suprised
by these deliveries which are apparently appearing outside of the scope
of any subscription it knows about. This really seems to be a bug in the
spec rather than our implementation or the client.

At the very least, the basic.recover_ok should contain a new ctag (which
for symmetry suggests the client should be able to specify one in
basic.recover to mirror basic.consume and basic.consume_ok) which the
client can then use to identify these deliveries and route them
accordingly.

> I was digging into the source code after sending my email and I found that (correct me if I'm wrong, it's my first time
> I look into this code) in rabbit_channel.erl:578 the recover is taking the unack'd messages
> and redelivering these using the same ConsumerTag as it was sent before. I think this is
> indeed what the FIXME warns: "What should happen if the consumer's been cancelled since?".

Absolutely, and well found, you are exactly right.

I would recommend that you don't use recover without also setting
requeue=true, thus ensuring the messages are inserted back into the
queue and can hence be redelivered to any valid consumer again (at
which point the ctag will be set correctly). Hopefully you'll at
least get unsurprising behaviour! I will raise some bugs to cover the
issues uncovered - many thanks for the report.

Matthew