Persistent Session messages dropped during connect() due to race condition (Subscribers: 0)

[ricardodanielalbano]

Description:

I encountered a critical issue when using Persistent Sessions (Clean Session = false) with QoS 1.
When the worker connects to the broker, if there are queued offline messages, the broker sends them immediately after the CONNACK. The library reads and processes these PUBLISH packets inside the execution of the $client->connect() method.
Since the standard usage pattern is:
$client->connect(...)
$client->subscribe(...)
The messages are processed at step 1. At this point, the subscription callback (step 2) has not been registered yet. The logger shows Subscribers: 0 and the messages are effectively dropped/lost.

Reproduction Steps:

Connect a client with CleanSession = false and QoS 1.
Disconnect the client.
Publish a message to a topic with QoS 1.
Reconnect the client using the standard flow (connect then subscribe).
Result: The offline message is consumed by the library during the connection handshake but not delivered to the application because the subscription wasn't registered in time.

Expected Behavior:

The library should either:
A. Buffer incoming PUBLISH packets received during the connect() phase and process them only after the connection flow returns control to the user.
B. Or allow registering subscriptions/callbacks before connecting (without needing to manually inject a MemoryRepository).

Current Workaround:

I had to manually instantiate a MemoryRepository, add the Subscription object to it, and inject this repository into the MqttClient constructor. This ensures the callback is present in memory before the socket is read.
While the workaround works, the default usage pattern failing silently for persistent sessions seems like a significant pitfall.

[Namoshek]

It's a known issue with the memory repository. The real error is transmitting the acknowledgent when not delivering the message to a subscription. I'll look into it. Thanks for bringing it up!

[Namoshek]

So I've been looking into this a bit and it turns out that a fix would actually be a breaking change, so I'm a bit hesitant to just make the change. The affected code is:

client/src/MqttClient.php

Lines 707 to 741 in 163c5b8

	protected function handleMessage(Message $message): void
	{
	// PUBLISH (incoming)
	if ($message->getType()->equals(MessageType::PUBLISH())) {
	if ($message->getQualityOfService() === self::QOS_AT_LEAST_ONCE) {
	// QoS 1.
	$this->sendPublishAcknowledgement($message->getMessageId());
	}
	if ($message->getQualityOfService() === self::QOS_EXACTLY_ONCE) {
	// QoS 2, part 1.
	try {
	$pendingMessage = new PublishedMessage(
	$message->getMessageId(),
	$message->getTopic(),
	$message->getContent(),
	2,
	$message->getRetained()
	);
	$this->repository->addPendingIncomingMessage($pendingMessage);
	} catch (PendingMessageAlreadyExistsException) {
	// We already received and processed this message.
	}
	// Always acknowledge, even if we received it multiple times.
	$this->sendPublishReceived($message->getMessageId());
	// We only deliver this published message as soon as we receive a publish complete.
	return;
	}
	// For QoS 0 and QoS 1 we can deliver right away.
	$this->deliverPublishedMessage($message->getTopic(), $message->getContent(), $message->getQualityOfService(), $message->getRetained());
	return;
	}

The solution would be to move L711-713 after L739 and add a condition to check if the message was actually delivered to a subscription. But in order for this to have any meaningful effect, the deliverPublishedMethod (which is currently protected and hence overridable) would need to return whether the message was delivered to a subscription or not. However, I don't really want to introduce a breaking change for a quite rare issue.

I think the better solution would be to implement a custom repository like you did, or to use a message received event handler that is called regardless of whether a subscription was served or not (L927):

client/src/MqttClient.php

Lines 900 to 928 in 163c5b8

	protected function deliverPublishedMessage(string $topic, string $message, int $qualityOfServiceLevel, bool $retained = false): void
	{
	$subscribers = $this->repository->getSubscriptionsMatchingTopic($topic);
	$this->logger->debug('Delivering message received on topic [{topic}] with QoS [{qos}] from the broker to [{subscribers}] subscribers.', [
	'topic' => $topic,
	'message' => $message,
	'qos' => $qualityOfServiceLevel,
	'subscribers' => count($subscribers),
	]);
	foreach ($subscribers as $subscriber) {
	if ($subscriber->getCallback() === null) {
	continue;
	}
	try {
	call_user_func($subscriber->getCallback(), $topic, $message, $retained, $subscriber->getMatchedWildcards($topic));
	} catch (\Throwable $e) {
	$this->logger->error('Subscriber callback threw exception for published message on topic [{topic}].', [
	'topic' => $topic,
	'message' => $message,
	'exception' => $e,
	]);
	}
	}
	$this->runMessageReceivedEventHandlers($topic, $message, $qualityOfServiceLevel, $retained);
	}