queue deliver different expiration messages to dead letter route at the same time

[swellee]

Describe the bug

with rabbitmq version 3.10.17,
when i set dead-letter-exchange and dead-letter-routing-key for queue A,
and I publish 3 messages to queue A with vary expirations of 3000, 10000, 60000,
so the first message was supposed to be delivered to dead-letter queue after 3 seconds,
and the second should be deliver to dead-letter queue after 10 seconds, and the 3rd should be 60s.
while result is, I can't get the dead-letters one by one,
but after 60s, those 3 messages were delivered to dead-letter-routed queue once at same time.

Reproduction steps

1. rabbitmq ver 3.10.17
2. setup exchange EX
3. setup queue A, and bind to exchange EX, route 'a' , config dead-letter-exchange to 'EX', dead-letter-routing-key to 'b';
   4.setup queue B, bind to exchange EX, route 'b';
4. publish msg 1 to route 'a' with expiration=3000, publish msg 2 to route 'a' with expiration=10000, publish msg 3 to route 'a' with expiration=60000
5. consume queue 'b',
6. watch the log
   ...

Expected behavior

should get msg1, msg2, msg3 one by one at their specificed expiration time.

Additional context

No response

[mkuratczyk]

Please provide an executable way of reproducing the problem. You can use whatever programming language you are comfortable with, including shell (rabbitmqadmin and/or https://perftest.rabbitmq.com/ can be used for this).

Based on the symptoms, it sounds like you are publishing messages with longer expiration first.

[michaelklishin]

Messages are expired once they reach the head of the queue. Until they do, they will not be dead-lettered. This is documented in the TTL guide and has been the case for years. In other words, expired messages do not "jump to the head of the queue". This is intentional.

Even though they are not marked for deletion and dead lettered before reaching the head of the queue, consumers are guaranteed to never have these messages delivered.

Neither dead lettering nor expiration were designed to be used as a precise retry-after-a-period-of-time mechanism.

[swellee]

hey, @michaelklishin , @mkuratczyk , thanks for your information.
actually, I want to use rabbitmq to implement schedule(delayed) msg based on the dead letter feature.
here's a brief code in nodejs, with npm package "amqp-connection-manager, amqplib":

`import amqp from "amqp-connection-manager";
/**

• brief: use dead-letter feature to implement schedule msg

*/

const queue_schedule = "queue_a";
const config = {
host: "amqp://localhost:5671",
username: "test",
password: "1234",

exchange: "EX",
scheduleRoute: "schedule_route",
scheduleRouteDLX: "schedule_route_dlx",
};
let channel;

async function main() {
const [protocol, host] = config.host.split("://");
const [hostname, port] = host.split(":");
const connection = await amqp.connect(
{
protocol,
hostname,
port: +port,
username: config.username,
password: config.password,
},
{
reconnectTimeInSeconds: 3,
heartbeatIntervalInSeconds: 10,
}
);

connection.on("connect", () => {
console.log("MQ Connected!");
});
connection.on("disconnect", (err) => {
console.log("MQ Disconnected.", err);
});
connection.on("connectFailed", (err) => {
console.log("MQ ConnectFailed.", err);
});

// init channel
channel = await connection.createChannel();
// init exchange
await channel.assertExchange(config.exchange, "direct", { durable: true });
// init _dlx queue first
const queue_schedule_dlx = queue_schedule + "_dlx";
await channel.addSetup(async (channel) => {
await channel.assertQueue(queue_schedule_dlx, {
durable: true,
exclusive: false,
arguments: { "x-queue-type": "classic" },
messageTtl: 1000 * 60 * 3,
});
await channel.bindQueue(
queue_schedule_dlx,
config.exchange,
config.scheduleRouteDLX
);
});
channel.consume(queue_schedule_dlx, onScheduleMsg);

//init schedule queue and set dlx to schedule_dlx
await channel.addSetup(async (channel) => {
await channel.assertQueue(queue_schedule, {
durable: true,
exclusive: false,
arguments: { "x-queue-type": "classic" },
deadLetterExchange: config.exchange,
deadLetterRoutingKey: config.scheduleRouteDLX,
});
await channel.bindQueue(
queue_schedule,
config.exchange,
config.scheduleRoute
);
});

await channel.waitForConnect();
console.log("MQ setup completed.");
}

function onScheduleMsg(msg) {
if (msg?.content) {
let content;
try {
content = JSON.parse(Buffer.from(msg.content).toString());
} catch (error) {
console.log("mq:parse message content failed, ark and skip");
this.channel.ack(msg, true);
return;
}
console.log("[GOT-MQ-SCHEDULE]", content);
this.channel.ack(msg, true);
}
}

/*
@parameter content type of ScheduleSetting
*/
async function sendScheduleMsg(content) {
if (!this.config?.host) {
console.log("mq init failed, skip send msg", content);
}
const { expiration, ...data } = content;
const buff = Buffer.from(JSON.stringify(data));
if (expiration < 1000) {
// treat as immediate msg
this.onScheduleMsg({ content: buff, fields: null, properties: null });
return;
}
console.log("mq:send-schedule-msg", content, expiration);
this.channel.sendToQueue(queue_schedule, buff, {
persistent: true,
expiration,
});
}

async function test() {
const times = [3000, 10000, 60000]
for (const expiration of times) {
await sendScheduleMsg({
expiration,
date: Date.toString(),
})
}
}
//-------------------test--------------
await main();
test();
`

[swellee]

seems like each queue's message-ttl is the MAX of (queue's message-ttl, message1-ttl, message2-ttl, ....), right?
if so, I think I should use rabbitMq's rabbitmq_delayed_message_exchange plugin.

[mkuratczyk]

https://www.rabbitmq.com/ttl.html#per-message-ttl-in-publishers
When both a per-queue and a per-message TTL are specified, the lower value between the two will be chosen.