If the Received event occurs before "reading e.Body.ToArray()", then the message is incorrect

[quanzhenying]

Describe the bug

Messages from different queues are interweaving across threads.
Affected versions: All versions of 6.x

`
Received Event

  await Task.Delay(rd.Next(0, 2));
  var project = Encoding.UTF8.GetString((byte[])msg.BasicProperties.Headers["project"]);
  var body = msg.Body.ToArray();
  string message = Encoding.UTF8.GetString(body);
  Console.WriteLine($"Received project/body: {project}/{message}");

`

Reproduction steps

1.Two queues are consuming messages simultaneously, using the same "Received" event for processing.
2.Within the Received event, add an await before reading the Body.
3.If the line await Task.Delay(rd.Next(0, 2)); is removed, the issue will not reproduce.

All code:
`
using RabbitMQ.Client.Events;
using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory
{
    HostName = "localhost",
    Port = 5672,
    UserName = "admin",
    Password = "123456"
};

var rd = new Random();
using IConnection connection = factory.CreateConnection();
var exchange = "iot.exchange";

//publish data
using var producer = connection.CreateModel();
producer.ExchangeDeclare(exchange, "headers", true, false);
var arguments = new Dictionary<string, object>
{
    { "x-queue-type", "quorum" }
};
producer.QueueDeclare(queue: "1", durable: true, exclusive: false, autoDelete: false, arguments);
producer.QueueBind("1", exchange, "", new Dictionary<string, object>
{
    { "project", "1" },
    { "x-match", "all" }
});
producer.QueueDeclare(queue: "2", durable: true, exclusive: false, autoDelete: false, arguments);
producer.QueueBind("2", exchange, "", new Dictionary<string, object>
{
    { "project", "2" },
    { "x-match", "all" }
});
Enumerable.Range(0, 200).ToList().ForEach(i =>
{
    var props = producer.CreateBasicProperties();
    props.ContentType = "application/octet-stream";
    props.DeliveryMode = 2;
    props.Headers = new Dictionary<string, object>
    {
        { "project", "1" }
    };
    producer.BasicPublish(exchange, "", props, Encoding.UTF8.GetBytes("1"));

    props = producer.CreateBasicProperties();
    props.ContentType = "application/octet-stream";
    props.DeliveryMode = 2;
    props.Headers = new Dictionary<string, object>
    {
        { "project", "2" }
    };
    producer.BasicPublish(exchange, "", props, Encoding.UTF8.GetBytes("2"));
});

//consumer
string[] queueNames = { "1", "2" };
foreach (var queuename in queueNames)
{
    var channel = connection.CreateModel();
    // set channel.BasicQos
    channel.BasicQos(0, 1, false);

    var consumer = new EventingBasicConsumer(channel);
    channel.ConfirmSelect();
    consumer.Received += async (model, msg) =>
    {
        //init tasks / eq. dbContext
        await Task.Delay(rd.Next(0, 2));

        var project = Encoding.UTF8.GetString((byte[])msg.BasicProperties.Headers["project"]);
        var body = msg.Body.ToArray();
        string message = Encoding.UTF8.GetString(body);
        Console.WriteLine($"Received project/body: {project}/{message}");

        if (project != message)
            Console.WriteLine("error: The body does not belong to this message");

        channel.BasicAck(msg.DeliveryTag, false);
    };
    channel.BasicConsume(queuename, false, consumer);
}

Console.WriteLine("Waiting for messages. To exit press CTRL+C");
Console.ReadLine();

`

Expected behavior

"Task.Delay(rd.Next(0, 2))" should not cause the Body to contain the content of another message.

Additional context

1.Version 5.1.2 does not have this issue;
2.Using “Task.Delay(rd.Next(0, 2)).GetAwaiter().GetResult()” also does not have this issue.

[michaelklishin]

@quanzhenying the concurrent dispatch behavior is not a bug. This is a perfectly fine behavior given that you consume from N queues concurrently on N channels. Delivery across channels are not coordinated in any way, and this never was the goal.

The fact that you did not observe it on 5.x is a coincidence, a difference in consumer operation dispatch implementations.

As for message body operations, see Consumer Memory Safety which is a new behavior by design in 6.x and onwards.