DisconnectedAsync is not raised if (failed) Publish attempts were made

[StefanLochauMF]

Describe the bug

For my use case, I have a two-way application protocol, where I periodically send a message to a different MQTT client.
This seems to lead to MqttClient.DisconnectedAsync to never be invoked by the KeepAlive mechanism when the connection has been disrupted.

Which component is your bug related to?

• Client

To Reproduce

Steps to reproduce the behavior:

1. Using this version of MQTTnet v5.1.0.1559.
2. Connect an instance of MqttClient to a broker using a KeepAliveInteval of 5 seconds.
3. Start to publish messages once every second.
4. Stop the broker.
5. Wait for the DisconnectedAsync event to be raised.

Expected behavior

The event is raised after at most 10s.

Actual behavior

The event is never raised.

Code example

// Using Testcontainers.Mosquitto v4.10.0 to create a broker
await using var mosquitto = new MosquittoBuilder("eclipse-mosquitto:2.1.2-alpine")
    .WithPortBinding(1883, assignRandomHostPort: true)
    .Build();
await mosquitto.StartAsync();

using var client = new MqttClientFactory().CreateMqttClient();
await client.ConnectAsync(
    new MqttClientOptionsBuilder()
        .WithConnectionUri(mosquitto.GetConnectionString())
        .WithKeepAlivePeriod(TimeSpan.FromSeconds(5))
        .Build());

var disconnectedRaised = new TaskCompletionSource();
client.DisconnectedAsync += _ =>
{
    disconnectedRaised.TrySetResult();
    return Task.CompletedTask;
};


using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
var periodicPublishTask = Task.Run(async () =>
{
    using var periodicTimer = new PeriodicTimer(TimeSpan.FromSeconds(1));
    await using var cancelTimer = cts.Token.Register(periodicTimer.Dispose);

    while (await periodicTimer.WaitForNextTickAsync(CancellationToken.None))
    {
        // When commenting this line, the keep alive mechanism does raise DisconnectedAsync event
        await client.PublishStringAsync("any/topic", "Any message");
    }
});

try
{
    await mosquitto.PauseAsync(CancellationToken.None);

    // This should definitely complete within the 30s, but times out
    await disconnectedRaised.Task.WaitAsync(cts.Token);
}
finally
{
    await cts.CancelAsync();
    await periodicPublishTask;
}