Error DataTransferException code 65

[Benjamin-Velluet]

Hello everyone,

In my project up to 1 time every 24 hours I come across this error:

[2022-05-04 22:58:49] production.ERROR: [65] Transferring data over socket failed: Sending data over the socket failed. Has it been closed? {"exception":"[object] (PhpMqtt\\Client\\Exceptions\\DataTransferException(code: 65): [65] Transferring data over socket failed: Sending data over the socket failed. Has it been closed? at /home/forge/livai.testsclaralockers.com/vendor/php-mqtt/client/src/MqttClient.php:1030)
[stacktrace]
#0 /home/forge/livai.testsclaralockers.com/vendor/php-mqtt/client/src/MqttClient.php(994): PhpMqtt\\Client\\MqttClient->writeToSocket()
#1 /home/forge/livai.testsclaralockers.com/vendor/php-mqtt/client/src/MqttClient.php(603): PhpMqtt\\Client\\MqttClient->ping()
#2 /home/forge/livai.testsclaralockers.com/app/Services/Mqtt/Mqtt.php(66): PhpMqtt\\Client\\MqttClient->loop()
#3 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Support/Facades/Facade.php(261): App\\Services\\Mqtt\\Mqtt->loop()
#4 /home/forge/livai.testsclaralockers.com/routes/console.php(36): Illuminate\\Support\\Facades\\Facade::__callStatic()
#5 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(36): Illuminate\\Foundation\\Console\\ClosureCommand->{closure}()
#6 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Container/Util.php(40): Illuminate\\Container\\BoundMethod::Illuminate\\Container\\{closure}()
#7 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(81): Illuminate\\Container\\Util::unwrapIfClosure()
#8 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Container/BoundMethod.php(37): Illuminate\\Container\\BoundMethod::callBoundMethod()
#9 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Container/Container.php(651): Illuminate\\Container\\BoundMethod::call()
#10 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Foundation/Console/ClosureCommand.php(55): Illuminate\\Container\\Container->call()
#11 /home/forge/livai.testsclaralockers.com/vendor/symfony/console/Command/Command.php(299): Illuminate\\Foundation\\Console\\ClosureCommand->execute()
#12 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Console/Command.php(121): Symfony\\Component\\Console\\Command\\Command->run()
#13 /home/forge/livai.testsclaralockers.com/vendor/symfony/console/Application.php(978): Illuminate\\Console\\Command->run()
#14 /home/forge/livai.testsclaralockers.com/vendor/symfony/console/Application.php(295): Symfony\\Component\\Console\\Application->doRunCommand()
#15 /home/forge/livai.testsclaralockers.com/vendor/symfony/console/Application.php(167): Symfony\\Component\\Console\\Application->doRun()
#16 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Console/Application.php(92): Symfony\\Component\\Console\\Application->run()
#17 /home/forge/livai.testsclaralockers.com/vendor/laravel/framework/src/Illuminate/Foundation/Console/Kernel.php(129): Illuminate\\Console\\Application->run()
#18 /home/forge/livai.testsclaralockers.com/artisan(37): Illuminate\\Foundation\\Console\\Kernel->handle()
#19 {main}
"} 

Normally the connection has no reason to be cut. My mqtt broker is an IOT scaleway and the code is on a digital ocean server.
I'm starting to run out of ideas, do you know where this can come from ?

console.php

include_once "toolsConsole.php";

Artisan::command('listen', function () {
    $subscriptionOpen = [];

    MqttListen::subscribe('livai/listen', function (string $topic, string $message) use (&$subscriptionOpen) {
        if ($message === 'restart') {
            foreach ($subscriptionOpen as $subscription) {
                MqttListen::unsubscribe($subscription);
            }
            $subscriptionOpen = subscribe();
        }
    });

    $subscriptionOpen = subscribe();

    MqttListen::loop();
});

toolsConsole.php

const CHANNELS = [
    '/locker/status',
    '/locker/status/module',
    '/locker/number/module',
    '/box/unlock',
    '/box/status'
];

function subscribe(): array
{
    $subscriptionOpen = [];

    foreach (Locker::all() as $locker) {
        if ($locker->mqttClientId) {
            foreach (CHANNELS as $channel) {
                MqttListen::subscribe('lockers/' . $locker->mqttClientId . '/return' . $channel, function (string $topic, string $message) {
                    $message = json_decode($message, true) ?: [];
                    if (!isset($message['requestId']) || !$request = Request::get($message['requestId'])) {
                        Log::error('Listen - No requestId or not found', [
                            'Class' => 'routes/console.php',
                            'Code' => 0,
                            'Time' => Carbon::now()->toISOString()
                        ]);
                    } else {
                        $request->update([
                            'receiveChannel' => $topic,
                            'return' => json_encode($message)
                        ]);
                    }
                });

                $subscriptionOpen[] = 'lockers/' . $locker->mqttClientId . '/return' . $channel;
            }

            MqttListen::subscribe('lockers/' . $locker->mqttClientId . '/webhook', function (string $topic, string $message) use ($locker) {
                foreach ($locker->consumers as $consumer) {
                    dispatch(new Webhook($consumer, [
                        'lockerId' => $locker->id,
                        'type' => 'LOCK',
                        'body' => json_decode($message, true)['body']
                    ]));
                }
            });

            MqttListen::subscribe('lockers/' . $locker->mqttClientId . '/log', function (string $topic, string $message) {
                $message = json_decode($message, true) ?: [];
                if (!isset($message['requestId']) || !$request = Request::get($message['requestId'])) {
                    Log::error('Listen - No requestId or not found', [
                        'Class' => 'routes/console.php',
                        'Code' => 0,
                        'Time' => Carbon::now()->toISOString()
                    ]);
                } else {
                    $request->update([
                        'log' => json_encode($message)
                    ]);
                }
            });

            $subscriptionOpen[] = 'lockers/' . $locker->mqttClientId . '/log';
        }
    }

    return $subscriptionOpen;
}

[Namoshek]

Does the issue occur at about the same time of day or is it random?

[Benjamin-Velluet]

Hi Namoshek,

The problem occurs randomly

[Namoshek]

From what I can see, you are not adding the lockers/{clientId}/webhook topic to your $subscriptionOpen[] and therefore no unsubscribe() is called before resubscribing. In my opinion it shouldn't cause an issue on the broker, but it might cause an issue on the client as the callback is registered again and memory exhaustion could be the result.

---

As a side note, I would recomment working a bit more with early exit to reduce indentation and complexity:

function subscribe(): array
{
    $subscriptionOpen = [];

    foreach (Locker::all() as $locker) {
        if (blank($locker->mqttClientId)) {
            continue;
        }
            
        // ... other code ...
    }

    return $subscriptionOpen;
}

You might also want to use return instead of if-else and avoid re-encoding the $message by using $decodedMessage instead:

MqttListen::subscribe('lockers/' . $locker->mqttClientId . '/return' . $channel, function (string $topic, string $message) {
    $decodedMessage = json_decode($message, true) ?: [];

    if (!isset($decodedMessage['requestId']) || !($request = Request::get($decodedMessage['requestId']))) {
        Log::error('Listen - No requestId or not found', [
            'Class' => 'routes/console.php',
            'Code' => 0,
            'Time' => Carbon::now()->toISOString()
        ]);
        
        return;
    } 

    $request->update([
        'receiveChannel' => $topic,
        'return' => $message
    ]);
});

[Namoshek]

By the way, are you using the mqtt-3.1.1 branch or are you connecting to Scaleway through MQTT 3.1?

[Benjamin-Velluet]

Indeed my processor is supposed to loop on the same subjects without ever stopping, it must simply send the return mqtt on another channel.

The code should never stop.

Is it a problem to subscribe indefinitely to several topics?

[Benjamin-Velluet]

Sorry I just realized I was actually using php-mqtt/laravel-client v1.0.0

Do you want me to re-edit the topic in the correct repository ?

[Namoshek]

Is it a problem to subscribe indefinitely to several topics?

No and that is not what I meant. Your resubscribe logic currently does not unsubscribe all topics before resubscribing them again.

Sorry I just realized I was actually using php-mqtt/laravel-client v1.0.0

Do you want me to re-edit the topic in the correct repository ?

No, that's fine. The problem most likely is in this library if at all anyway. My question was aiming at Scaleway support for MQTT 3.1 since the website says they offer MQTT 3.1.1 (which has a different protocol version).

[Benjamin-Velluet]

No and that is not what I meant. Your resubscribe logic currently does not unsubscribe all topics before resubscribing them again.

Yes indeed, error on my part, I will correct that. I'm not using this feature yet, though, so there's no reason for it error to come from here.

No, that's fine. The problem most likely is in this library if at all anyway. My question was aiming at Scaleway support for MQTT 3.1 since the website says they offer MQTT 3.1.1 (which has a different protocol version).

Ok I understand, indeed Scaleway use MQTT 3.1.1, do you think the problem comes from here?

[Namoshek]

Ok I understand, indeed Scaleway use MQTT 3.1.1, do you think the problem comes from here?

I'm actually quite surprised that the client connects to the Scaleway broker at all since MQTT 3.1.1 uses an entirely different header than MQTT 3.1. You might want to check out PR #90 (and look for getEncodedProtocolNameAndVersion() specifically) to see the difference. So I'm actually quite certain that Scaleway has at least some support for MQTT 3.1 built in as well.

If you were using the resubscribe logic, I could imaging the lines 808-816 of PR 90 (which is missing in the current version) to be an isssue. But if you don't, the only thing which comes to my mind is that the disconnect comes from the platform itself. Obviouly they do need to restart their services once in a while to deploy updates. Do you have a way to ask their support for more details regarding the disconnects and/or whether they disconnect clients during updates?

And maybe another question: how much data / how many messages are you currently processing per time unit (seconds/minutes/...)?

[Benjamin-Velluet]

If you were using the resubscribe logic, I could imaging the lines 808-816 of PR 90 (which is missing in the current version) to be an isssue. But if you don't, the only thing which comes to my mind is that the disconnect comes from the platform itself. Obviouly they do need to restart their services once in a while to deploy updates. Do you have a way to ask their support for more details regarding the disconnects and/or whether they disconnect clients during updates?

I asked them, there are no connection cuts on their side, even in the event of a server restart.

And maybe another question: how much data / how many messages are you currently processing per time unit (seconds/minutes/...)?

I communicate with MQTT every 5 minutes at most, I do not send more than 10/15 messages per minute

[Benjamin-Velluet]

Ok, it's now more than 48 hours since the error has not returned. Here is my latest commit.

When I call MQTT:: in fact it was this file that I call :

Mqtt.php

<?php

namespace App\Services\Mqtt;

use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\MqttClient;

class Mqtt
{
    protected $mqtt;

    protected function settings(string $pathCaObject, string $pathCaKeyObject, string $pathCaServer): ConnectionSettings
    {
        return (new ConnectionSettings)
            ->setConnectTimeout(env('MQTT_CONNECT_TIMEOUT', 60))
            ->setSocketTimeout(env('MQTT_SOCKET_TIMEOUT', 5))
            ->setResendTimeout(env('MQTT_RESEND_TIMEOUT', 10))
            ->setKeepAliveInterval(env('MQTT_KEEP_ALIVE_INTERVAL', 10))
            ->setUsername(env('MQTT_AUTH_USERNAME'))
            ->setPassword(env('MQTT_AUTH_PASSWORD'))
            ->setUseTls(env('MQTT_TLS_ENABLED', false))
            ->setTlsSelfSignedAllowed(env('MQTT_TLS_ALLOW_SELF_SIGNED_CERTIFICATE', false))
            ->setTlsVerifyPeer(env('MQTT_TLS_VERIFY_PEER', true))
            ->setTlsVerifyPeerName(env('MQTT_TLS_VERIFY_PEER_NAME', true))
            ->setTlsCertificateAuthorityFile($pathCaServer)
            ->setTlsCertificateAuthorityPath(env('MQTT_TLS_CA_PATH'))
            ->setTlsClientCertificateFile($pathCaObject)
            ->setTlsClientCertificateKeyFile($pathCaKeyObject)
            ->setTlsClientCertificateKeyPassphrase(env('MQTT_TLS_CLIENT_CERTIFICATE_KEY_PASSPHRASE'))
            ->setLastWillTopic(env('MQTT_LAST_WILL_TOPIC'))
            ->setLastWillMessage(env('MQTT_LAST_WILL_MESSAGE'))
            ->setLastWillQualityOfService(env('MQTT_LAST_WILL_QUALITY_OF_SERVICE', 2))
            ->setRetainLastWill(env('MQTT_LAST_WILL_RETAIN', false));
    }


    public function __construct()
    {
        $this->mqtt = new MqttClient(env('MQTT_HOST'), env('MQTT_PORT'));
        $this->mqtt->connect($this->settings(env('MQTT_TLS_CLIENT_CERT_FILE'), env('MQTT_TLS_CLIENT_CERT_KEY_FILE'), env('MQTT_TLS_CA_FILE')));
    }

    public function loop(bool $allowSleep = true, bool $exitWhenQueuesEmpty = false, int $queueWaitLimit = null): void
    {
        $this->mqtt->loop($allowSleep, $exitWhenQueuesEmpty, $queueWaitLimit);
    }

    public function subscribe(string $topicFilter, callable $callback = null, int $qualityOfService = 2): void
    {
        $this->mqtt->subscribe($topicFilter, $callback, $qualityOfService);
    }

    public function publish(string $topic, string $message, int $qualityOfService = 2, bool $retain = false): void
    {
        $this->mqtt->publish($topic, $message, $qualityOfService, $retain);
        if ($qualityOfService) {
            $this->loop(true, true);
        }
    }

    public function disconnect(): void
    {
        if ($this->mqtt->isConnected()) {
            $this->mqtt->disconnect();
        }
    }
}

But here are the modifications of my last commit which we stop the error :

public function publish(string $topic, string $message, int $qualityOfService = 0, bool $retain = false): void
    {
        $this->mqtt->publish($topic, $message, $qualityOfService, $retain);
    }

I made this modification because it blocked in the loop after publish

[Namoshek]

Are you publishing within the callback of a subscription (using QoS > 0)? If so, you don't have to (and also shouldn't) start a second loop since other messages received for subscriptions will be delivered in the same context, potentially causing an infinite depth of callbacks through recursion which can cause a stack overflow and memory exhaustion.

Otherwise, the code looks fine. Only thing I noticed: if you are using it within Laravel, you should not access env() directly but through config() to avoid issues when caching the config.

[domainfun]

It's an interesting discussion, thanks for all your comments @Namoshek.

Transferring data over socket failed: Sending data over the socket failed. Has it been closed?

The ticket started with the above error message being reported. We see this message too, at irregular intervals, but don't expect that it's a bug in the MQTT client.

In our case, we currently run two MQTT clients from servers hosted at two different UK-based hosting companies, and both MQTT clients connect to the Amazon MQTT IoT service.

The frequency/pattern for "Sending data over the socket failed" occurring varies greatly:

1. randomly once or multiple times a day for several days (but then all is fine for weeks),
2. randomly once a week/fortnight/month,
3. for both clients at the same time once in a while.

I understood that this is mainly caused by the stability of the Internet connection between our hosting companies where we have the MQTT client and the Amazon IoT server, and point 3 above was clearly when Amazon restarted their services.

Our developer commented in August 2021:

The MqttClient library does not have the ability to check if the connection is still open (there is a public method isConnected() but this only returns the state of the object not the socket's actual state).
However, by capturing the exception thrown by the MqttClient the our_application:iot:process method will now attempt to re-connect with the broker with a new clean session. Any failed message will then be reprocessed in the next event loop.

Perhaps the above helps @Benjamin-Velluet to point him in the right direction for improving his code?

As for this MQTT client library, is it possible (& sensible) to consider adding some code which would re-establish a socket connection when the socket has been unexpectedly closed? I suppose this would need additional library settings like "max number of retries" and "max duration of retries", with sensible defaults?

Our current plan is to monitor the connection made by the MQTT client with something like Supervisor, as we're running a CLI script for this as part of our Symfony web application, and would ask Supervisor to restart the CLI script when the socket connection has been closed unexpectedly (as our CLI script will terminate when such an error occurs).

Does the above all make sense and helps to clarify what's happening here?

[Namoshek]

It's an interesting discussion, thanks for all your comments @Namoshek.

It is interesting for me as well and to be honest, it is exactly what I'm looking for in open source. 👍

Perhaps the above helps @Benjamin-Velluet to point him in the right direction for improving his code?

As for this MQTT client library, is it possible (& sensible) to consider adding some code which would re-establish a socket connection when the socket has been unexpectedly closed? I suppose this would need additional library settings like "max number of retries" and "max duration of retries", with sensible defaults?

I agree that this could be implemented in user-land code and I also agree that it would make more sense to integrate this into the library since it could be of help for many people. The reason this has not happened yet is that the behavior would need to depend on the connection settings and could become quite complex.

My primary concern is the clean session flag (description from the Mosquitto documentation):

Clean session / Durable connections
On connection, a client sets the "clean session" flag, which is sometimes also known as the "clean start" flag. If clean session is set to false, then the connection is treated as durable. This means that when the client disconnects, any subscriptions it has will remain and any subsequent QoS 1 or 2 messages will be stored until it connects again in the future. If clean session is true, then all subscriptions will be removed for the client when it disconnects.

In my opinion there are two options regarding an implementation:

• either disallow using the clean session flag in combination with automatic reconnects (my preferred solution)
• or use a different reconnect logic based on whether the clean session flag is set (difficult because it requires us to replay all subscriptions which requires async message flows and is problematic because of loop-in-loop scenarios)

How do you detect disconnects in your code @domainfun, are you simply catching and handling the DataTransferException?

Our current plan is to monitor the connection made by the MQTT client with something like Supervisor, as we're running a CLI script for this as part of our Symfony web application, and would ask Supervisor to restart the CLI script when the socket connection has been closed unexpectedly (as our CLI script will terminate when such an error occurs).

And yeah, that makes total sense and I would implement this behavior regardless of whether the MQTT client was able to auto-reconnect or not. No script is every going to run forever without supervision. 😄