How to know if Aws::Crt::Mqtt::MqttConnection::Publish fails? #594

nabelekt · 2024-01-25T04:33:24Z

nabelekt
Jan 25, 2024

Hello. The documentation states that Aws::Crt::Mqtt::MqttConnection::Publish will return:

packet id of the publish request, or 0 if the attempt failed synchronously

I have the following test code:

Code

#include <aws/crt/Api.h>
#include <aws/iot/MqttClient.h>
#include <unistd.h>  // For sleep
#include <iostream>  // For cout

int main()
{
    /************************ Setup ****************************/

    // Do the global initialization for the API.
    Aws::Crt::ApiHandle apiHandle;

    struct {
        Aws::Crt::String input_message   = "Hello World";
        Aws::Crt::String input_cert      = "/root/gateway/aws_keys/certificate.pem.crt";
        Aws::Crt::String input_key       = "/root/gateway/aws_keys/private.pem.key";
        Aws::Crt::String input_ca        = "/root/gateway/aws_keys/AmazonRootCA1.pem";
        Aws::Crt::String input_endpoint  = "a1ivqb4fk63pz7-ats.iot.us-east-1.amazonaws.com";
        Aws::Crt::String input_proxyHost = "";
        uint64_t input_proxyPort         = 0;
        uint64_t input_port              = 0;
        Aws::Crt::String input_clientId  = "basicPubSub";
        Aws::Crt::String input_topic     = "gateway/telemetry";
    } cmdData;

    // Create the MQTT builder and populate it with data from cmdData.
    auto clientConfigBuilder =
        Aws::Iot::MqttClientConnectionConfigBuilder(cmdData.input_cert.c_str(), cmdData.input_key.c_str());
    clientConfigBuilder.WithEndpoint(cmdData.input_endpoint);
    clientConfigBuilder.WithCertificateAuthority(cmdData.input_ca.c_str());

    // Create the MQTT connection from the MQTT builder
    auto clientConfig = clientConfigBuilder.Build();
    if (!clientConfig)
    {
        fprintf(
            stderr,
            "Client Configuration initialization failed with error %s\n",
            Aws::Crt::ErrorDebugString(clientConfig.LastError()));
        exit(-1);
    }
    Aws::Iot::MqttClient client = Aws::Iot::MqttClient();
    auto connection = client.NewConnection(clientConfig);
    if (!*connection)
    {
        fprintf(
            stderr,
            "MQTT Connection Creation failed with error %s\n",
            Aws::Crt::ErrorDebugString(connection->LastError()));
        exit(-1);
    }

    /**
     * In a real world application you probably don't want to enforce synchronous behavior
     * but this is a sample console application, so we'll just do that with a condition variable.
     */
    std::promise<bool> connectionCompletedPromise;
    std::promise<void> connectionClosedPromise;

    // Invoked when a MQTT connect has completed or failed
    auto onConnectionCompleted =
        [&](Aws::Crt::Mqtt::MqttConnection &, int errorCode, Aws::Crt::Mqtt::ReturnCode returnCode, bool)
        {
            if (errorCode)
            {
                fprintf(stdout, "Connection failed with error %s\n", Aws::Crt::ErrorDebugString(errorCode));
                connectionCompletedPromise.set_value(false);
            }
            else
            {
                fprintf(stdout, "Connection completed with return code %d\n", returnCode);
                connectionCompletedPromise.set_value(true);
            }
        };

    // Invoked when a MQTT connection was interrupted/lost
    auto onInterrupted = [&](Aws::Crt::Mqtt::MqttConnection &, int error)
    {
        fprintf(stdout, "Connection interrupted with error %s\n", Aws::Crt::ErrorDebugString(error));
    };

    // Invoked when a MQTT connection was interrupted/lost, but then reconnected successfully
    auto onResumed = [&](Aws::Crt::Mqtt::MqttConnection &, Aws::Crt::Mqtt::ReturnCode, bool)
    {
        fprintf(stdout, "Connection resumed\n");
    };

    // Invoked when a disconnect message has completed.
    auto onDisconnect = [&](Aws::Crt::Mqtt::MqttConnection &)
    {
        fprintf(stdout, "Disconnect completed\n");
        connectionClosedPromise.set_value();
    };

    // Assign callbacks
    connection->OnConnectionCompleted   = std::move(onConnectionCompleted);
    connection->OnDisconnect            = std::move(onDisconnect);
    connection->OnConnectionInterrupted = std::move(onInterrupted);
    connection->OnConnectionResumed     = std::move(onResumed);

    /************************ Run the sample ****************************/

    // Connect
    fprintf(stdout, "Connecting...\n");
    // keepAliveTimeSecs is the "time interval to space mqtt pings apart by."
    // https://aws.amazon.com/iot-core/pricing/additional-details/ says 30 s is the minimum to not be charged
    if (!connection->Connect(cmdData.input_clientId.c_str(), false /*cleanSession*/, 35 /*keepAliveTimeSecs*/))
    {
        fprintf(stderr, "MQTT Connection failed with error %s\n", Aws::Crt::ErrorDebugString(connection->LastError()));
        exit(-1);
    }

    if (connectionCompletedPromise.get_future().get())
    {
        uint32_t publishedCount = 0;

        while (publishedCount < 10000)
        {
            // printf("%s:%d\n", __FILE__, __LINE__);

            Aws::Crt::String messagePayload = "\"" + cmdData.input_message + " " + std::to_string(publishedCount).c_str() + "\"";
            Aws::Crt::ByteBuf payload = Aws::Crt::ByteBufFromArray((const uint8_t *)messagePayload.data(), messagePayload.length());

            auto onPublishComplete = [](Aws::Crt::Mqtt::MqttConnection &, uint16_t, int) {};
            uint16_t publishResult = connection->Publish(
                cmdData.input_topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, false, payload, onPublishComplete);
            printf("publishResult: %d\n", publishResult);
            publishedCount++;

            sleep(3);
        }

        // Disconnect
        if (connection->Disconnect())
        {
            connectionClosedPromise.get_future().wait();
        }
    }
    else
    {
        exit(-1);
    }

    return 0;
}

If I start it when the system does not have an Internet connection, I get the expected output:

root@GTWY-DEV-0003:~/gateway/build# ./aws_mqtt_publish.exe
Connecting...
Connection failed with error aws-c-io: AWS_IO_DNS_QUERY_FAILED, A query to dns failed to resolve.
root@GTWY-DEV-0003:~/gateway/build#

If I start it when the system does have an Internet connection, I still get the expected output:

root@GTWY-DEV-0003:~/gateway/build# ./aws_mqtt_publish.exe
Connecting...
Connection completed with return code 0
publishResult: 1
publishResult: 2
publishResult: 3
publishResult: 4
...

However, if I remove the Internet connection while that is running, Publish continues to return the packet ID as if the publish was successful, which it of course was not.

Whether by using the return value of Publish or another method, how can I tell if the Publish was actually successful?

Thank you!

xiazhvera · 2024-01-25T17:45:59Z

xiazhvera
Jan 25, 2024
Maintainer

The Publish() returns a packet id to track the publish, not the actual publish result. You would like to check the result in onPublishComplete. You can check the errorCode in OnOperationCompleteHandler to find the actual publish result.

0 replies

nabelekt · 2024-01-26T22:29:08Z

nabelekt
Jan 26, 2024
Author

Thank you for your support, @xiazhvera. I modified my test code so that I now have:

...
            auto onPublishComplete = [](Aws::Crt::Mqtt::MqttConnection& connection, uint16_t packetId, uint16_t errorCode)
            {
                printf("packetId, errorCode: %4d, %d\n", packetId, errorCode);
            };
...

This is now the output. I removed the Internet connection after packet 11:

Connecting...
Connection completed with return code 0
publishResult: 1
packetId, errorCode:    1, 0
publishResult: 2
packetId, errorCode:    2, 0
publishResult: 3
packetId, errorCode:    3, 0
publishResult: 4
packetId, errorCode:    4, 0
publishResult: 5
packetId, errorCode:    5, 0
publishResult: 6
packetId, errorCode:    6, 0
publishResult: 7
packetId, errorCode:    7, 0
publishResult: 8
packetId, errorCode:    8, 0
publishResult: 9
packetId, errorCode:    9, 0
publishResult: 10
packetId, errorCode:   10, 0
publishResult: 11
packetId, errorCode:   11, 0
publishResult: 12
publishResult: 13
publishResult: 14
publishResult: 15
publishResult: 16
publishResult: 17
publishResult: 18
publishResult: 19
publishResult: 20
publishResult: 21
publishResult: 22
publishResult: 24
Connection interrupted with error libaws-c-mqtt: AWS_ERROR_MQTT_TIMEOUT, Time limit between request and response has been exceeded.
publishResult: 25
publishResult: 26
publishResult: 27
publishResult: 28
publishResult: 29
publishResult: 30
publishResult: 31
publishResult: 32
publishResult: 33
publishResult: 34
publishResult: 35
publishResult: 36

So, it appears to me that the onPublishComplete lambda function is not running if the Publish fails. Am I still doing something incorrectly, or is this a bug?

1 reply

nabelekt Jan 26, 2024
Author

Maybe related to aws/aws-sdk-cpp#2654? Could it have to do with my qos argument (AWS_MQTT_QOS_AT_LEAST_ONCE)? Is there a long timeout for the Publish attempt that hasn't yet expired in my test so that onPublishComplete isn't getting called yet? In this test, there are three seconds between each packet publish attempt, so in the test above, I killed it ~75 seconds after the publish attempt for packet 12.

nabelekt · 2024-01-30T23:14:03Z

nabelekt
Jan 30, 2024
Author

@xiazhvera, should I file a bug report for this?

4 replies

bretambrose Jan 30, 2024
Maintainer

This isn't a bug. By default, there is no timeout on submitted publishes. You can set a client-wide operation timeout using this parameter at connect time: https://github.com/awslabs/aws-crt-cpp/blob/main/include/aws/crt/mqtt/MqttConnection.h#L257

The timeout only applies to the interval between a QoS1 publish going out on the wire and a corresponding ack being received. If the client is offline (which is the case in your example), then it will have no effect.

QoS 0 publishes are an exception and will be failed on submission if the client is offline; everything else will be queued.

We realize this is confusing but we cannot change the behavior outside of a major version bump which isn't going to happen anytime soon, unfortunately.

nabelekt Jan 31, 2024
Author

Thanks for your reply, @bretambrose. Are you saying that if I add a timeout in the Connect call, say 100 ms, and change AWS_MQTT_QOS_AT_LEAST_ONCE to AWS_MQTT_QOS_AT_MOST_ONCE in my Publish call, I should expect to see the onPublishComplete callback get called when the client is offline?

And if the publish fails, it will be queued until a connection is resumed? Where can I read more about that? Is there a way that I can check what is in the queue and a way to drop what is in the queue?

nabelekt Jan 31, 2024
Author

Edit: Removed

nabelekt Jan 31, 2024
Author

I made those changes to the Connect and Publish calls. This is my current test code:

Code


#include <aws/crt/Api.h>
#include <aws/iot/MqttClient.h>
#include <unistd.h>  // For sleep
#include <iostream>  // For cout

int main()
{
    /************************ Setup ****************************/

    // Do the global initialization for the API.
    Aws::Crt::ApiHandle apiHandle;

    struct {
        Aws::Crt::String input_message   = "Hello World";
        Aws::Crt::String input_cert      = "/root/gateway/aws_keys/certificate.pem.crt";
        Aws::Crt::String input_key       = "/root/gateway/aws_keys/private.pem.key";
        Aws::Crt::String input_ca        = "/root/gateway/aws_keys/AmazonRootCA1.pem";
        Aws::Crt::String input_endpoint  = "a1ivqb4fk63pz7-ats.iot.us-east-1.amazonaws.com";
        Aws::Crt::String input_proxyHost = "";
        uint64_t input_proxyPort         = 0;
        uint64_t input_port              = 0;
        Aws::Crt::String input_clientId  = "basicPubSub";
        Aws::Crt::String input_topic     = "gateway/telemetry";
    } cmdData;

    // Create the MQTT builder and populate it with data from cmdData.
    auto clientConfigBuilder =
        Aws::Iot::MqttClientConnectionConfigBuilder(cmdData.input_cert.c_str(), cmdData.input_key.c_str());
    clientConfigBuilder.WithEndpoint(cmdData.input_endpoint);
    clientConfigBuilder.WithCertificateAuthority(cmdData.input_ca.c_str());

    // Create the MQTT connection from the MQTT builder
    auto clientConfig = clientConfigBuilder.Build();
    if (!clientConfig)
    {
        fprintf(
            stderr,
            "Client Configuration initialization failed with error %s\n",
            Aws::Crt::ErrorDebugString(clientConfig.LastError()));
        exit(-1);
    }
    Aws::Iot::MqttClient client = Aws::Iot::MqttClient();
    auto connection = client.NewConnection(clientConfig);
    if (!*connection)
    {
        fprintf(
            stderr,
            "MQTT Connection Creation failed with error %s\n",
            Aws::Crt::ErrorDebugString(connection->LastError()));
        exit(-1);
    }

    /**
     * In a real world application you probably don't want to enforce synchronous behavior
     * but this is a sample console application, so we'll just do that with a condition variable.
     */
    std::promise<bool> connectionCompletedPromise;
    std::promise<void> connectionClosedPromise;

    // Invoked when a MQTT connect has completed or failed
    auto onConnectionCompleted =
        [&](Aws::Crt::Mqtt::MqttConnection &, int errorCode, Aws::Crt::Mqtt::ReturnCode returnCode, bool)
        {
            if (errorCode)
            {
                fprintf(stdout, "Connection failed with error %s\n", Aws::Crt::ErrorDebugString(errorCode));
                connectionCompletedPromise.set_value(false);
            }
            else
            {
                fprintf(stdout, "Connection completed with return code %d\n", returnCode);
                connectionCompletedPromise.set_value(true);
            }
        };

    // Invoked when a MQTT connection was interrupted/lost
    auto onInterrupted = [&](Aws::Crt::Mqtt::MqttConnection &, int error)
    {
        fprintf(stdout, "Connection interrupted with error %s\n", Aws::Crt::ErrorDebugString(error));
    };

    // Invoked when a MQTT connection was interrupted/lost, but then reconnected successfully
    auto onResumed = [&](Aws::Crt::Mqtt::MqttConnection &, Aws::Crt::Mqtt::ReturnCode, bool)
    {
        fprintf(stdout, "Connection resumed\n");
    };

    // Invoked when a disconnect message has completed.
    auto onDisconnect = [&](Aws::Crt::Mqtt::MqttConnection &)
    {
        fprintf(stdout, "Disconnect completed\n");
        connectionClosedPromise.set_value();
    };

    // Assign callbacks
    connection->OnConnectionCompleted   = std::move(onConnectionCompleted);
    connection->OnDisconnect            = std::move(onDisconnect);
    connection->OnConnectionInterrupted = std::move(onInterrupted);
    connection->OnConnectionResumed     = std::move(onResumed);

    /************************ Run the sample ****************************/

    // Connect
    fprintf(stdout, "Connecting...\n");
    // keepAliveTimeSecs is the "time interval to space mqtt pings apart by."
    // https://aws.amazon.com/iot-core/pricing/additional-details/ says 30 s is the minimum to not be charged
    if (!connection->Connect(cmdData.input_clientId.c_str(), /*cleanSession=*/false, /*keepAliveTimeSecs=*/35,
        /*pingTimeoutMs=*/0, /*protocolOperationTimeoutMs=*/100))
    {
        fprintf(stderr, "MQTT Connection failed with error %s\n", Aws::Crt::ErrorDebugString(connection->LastError()));
        exit(-1);
    }

    if (connectionCompletedPromise.get_future().get())
    {
        uint32_t publishedCount = 1;

        while (publishedCount < 10000)
        {
            // printf("%s:%d\n", __FILE__, __LINE__);

            Aws::Crt::String messagePayload = "\"" + cmdData.input_message + " " + std::to_string(publishedCount).c_str() + "\"";
            Aws::Crt::ByteBuf payload = Aws::Crt::ByteBufFromArray((const uint8_t *)messagePayload.data(), messagePayload.length());
            printf("\n%s -- messagePayload: %s\n", __FUNCTION__, messagePayload.c_str());

            auto onPublishComplete = [messagePayload](Aws::Crt::Mqtt::MqttConnection& connection, uint16_t packetId, uint16_t errorCode)
            {
                printf("%s -- packetId: %4d, messagePayload: %s, errorCode: %d\n", __FUNCTION__, packetId, messagePayload.c_str(), errorCode);
            };
            uint16_t publishResult = connection->Publish(
                cmdData.input_topic.c_str(), AWS_MQTT_QOS_AT_MOST_ONCE, false, payload, onPublishComplete);
            printf("%s -- publishResult: %d\n", __FUNCTION__, publishResult);
            publishedCount++;

            sleep(1);
        }

        // Disconnect
        if (connection->Disconnect())
        {
            connectionClosedPromise.get_future().wait();
        }
    }
    else
    {
        exit(-1);
    }

    return 0;
}

This is what I am seeing now: https://drive.google.com/file/d/1wu26MS-6_ku8NN6m4HX9plOXMmrOIiQL/view?usp=sharing

I start the program and things work as expected.
I remove the Internet connection.
The onPublishComplete callback continues to get called with errorCode 0 until about 20 seconds after the Internet connection was removed.
Then, I get the AWS_ERROR_MQTT_TIMEOUT error and onPublishComplete no longer gets called. publishResult is now 0 instead of the packetId.
This continues until I return the Internet connection.
The program continues to work as expected.

As you can see in the MQTT test client, messages 14-102 were dropped. In a previous test, it did seem like a few packets in that gap ended up getting published.

This is close to what we need. We'd actually prefer synchronous program execution where we call Publish and it does not return until the publish is successful or the specified timeout expires. And we could use the return value to determine if it was successful or not. We don't need asynchronous operation.

How can I shorten that ~20 seconds it takes for the API to realize there is no connection? I think I would set it to be less than protocolOperationTimeoutMs so that the packetId value returned will be 0. If that makes sense?
I did try setting both pingTimeoutMs and protocolOperationTimeoutMs to 500. Didn't seem to have much effect.

I have never seen errorCode passed to onPublishComplete be anything other than 0.

We don't need messages to be queued if the publish fails. We just need a way to determine if the publish fails so that we can try again later when the connection is reestablished.

bretambrose · 2024-01-31T18:38:02Z

bretambrose
Jan 31, 2024
Maintainer

In your latest example, you're using QoS 0. QoS 0 publish complete happens on socket write which is why you continue to get "successes" after severing the connectivity.

In general, use QoS 0 when it's not critical that a message get delivered -- transient state data, for example.
Use QoS 1 when you must have server-side acknowledgement that the publish was successfully transferred from client to broker.

Going back to a QoS 1 approach, if you're going to resubmit a failed publish later, could you let it sit in the client's queue until connectivity is re-established? You cant directly manipulate the queue, but you can query some basic properties (# of messages, approximate total bytes used) to guide decision-making.

The 5 client gives you a little more control over the queue by letting you establish a keep/reject policy:

aws-crt-cpp/source/mqtt/Mqtt5Client.cpp

Line 300 in c499dff

Mqtt5ClientOptions &Mqtt5ClientOptions::WithOfflineQueueBehavior(
https://github.com/awslabs/aws-c-mqtt/blob/171840f98dd628e60113c4033cc5b16a66079a7a/include/aws/mqtt/v5/mqtt5_client.h#L185-L215

There is no reliable way for the client to detect a disconnect outside of a full keep alive timeout, which for IoT Core is going to be ~45 seconds given a keep alive minimum of 30 seconds. Tcp keep alive might give you better results.

It sounds like you want an end-to-end timeout for publishes that, if reached, will fail the operation no matter what state it is in. We don't presently support that, but it might be a good potential feature. Such a feature would need to coexist nicely with the fact that certain MQTT operations must be retried on reconnect (any unacked QoS 1 publishes that have been written to the wire) by specification. So even if we did add such a feature there will still be QoS 1 publishes which cannot be failed by the end-to-end timeout.

In retrospect, the existing operation_timeout should have been "ack_timeout" leaving "timeout" available for a potential end-to-end feature.

1 reply

nabelekt Feb 3, 2024
Author

Yes, we want to publish a message, have the API ensure that it was received (presumedly by receiving some ACK from the server/broker, and have the publish fail if that doesn't happen.

We don't presently support that, but it might be a good potential feature.
Should I write a feature request issue? Would it be worked any time soon?

It is critical to us that we don't drop any data completely. Our ideas was to try and publish it, and if the publish fails for any reason, write it to a binary file for the publish to be re-attempted later.

Now I am thinking about implementing our own ACK. I think we could do the publish while being subscribed to a topic that could receive acknowledgments published by a Lambda function. Would certainly add complexity for us, so will think on it a bit more.

bretambrose · 2024-02-05T18:09:04Z

bretambrose
Feb 5, 2024
Maintainer

The unfortunate answer to the feature-request questions is that, at least right now, it's very unlikely in the short term. The team has planned work that reaches well into next year. Priorities change and we regularly shuffle things to react, but it wouldn't be a good idea to wait on the availability/expectation of such a feature.

I would advise against trying to re-create some kind of custom ack solution. Given what you've outlined so far, it seems like the following would work:

Before submitting a publish, write it to durable storage
On publish completion:
1. If successful, delete it from durable storage
1. If failed, do something appropriate (retry with backoff, or something else)
On application startup, read all durable storage entries and submit them to the client

2 replies

nabelekt Feb 5, 2024
Author

Thanks, @bretambrose. What you outlined is essentially my original plan. But, I am still stuck on numbers 3 and 4- how do I know if the publish was successful or not?

bretambrose Feb 5, 2024
Maintainer

The completion callback is authoritative. The fact that you aren't seeing it during extended outages means the client is still intending to attempt the publish as soon as connectivity is restored. Publishes are never "lost" or "disappear" without notice, with one notable exception that is beyond our control (if you exceed IoT Core's publish TPS limit for an extended period of time, the service front end will just start dropping incoming publishes as DOS protection; that behavior is what the ack timeout was originally introduced to handle).

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

How to know if Aws::Crt::Mqtt::MqttConnection::Publish fails? #594

{{title}}

Replies: 5 comments 8 replies

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{title}}

Select a reply

How to know if Aws::Crt::Mqtt::MqttConnection::Publish fails? #594

nabelekt Jan 25, 2024

Replies: 5 comments · 8 replies

xiazhvera Jan 25, 2024 Maintainer

nabelekt Jan 26, 2024 Author

nabelekt Jan 26, 2024 Author

nabelekt Jan 30, 2024 Author

bretambrose Jan 30, 2024 Maintainer

nabelekt Jan 31, 2024 Author

nabelekt Jan 31, 2024 Author

nabelekt Jan 31, 2024 Author

bretambrose Jan 31, 2024 Maintainer

nabelekt Feb 3, 2024 Author

bretambrose Feb 5, 2024 Maintainer

nabelekt Feb 5, 2024 Author

bretambrose Feb 5, 2024 Maintainer

nabelekt
Jan 25, 2024

Replies: 5 comments 8 replies

xiazhvera
Jan 25, 2024
Maintainer

nabelekt
Jan 26, 2024
Author

nabelekt Jan 26, 2024
Author

nabelekt
Jan 30, 2024
Author

bretambrose Jan 30, 2024
Maintainer

nabelekt Jan 31, 2024
Author

nabelekt Jan 31, 2024
Author

nabelekt Jan 31, 2024
Author

bretambrose
Jan 31, 2024
Maintainer

nabelekt Feb 3, 2024
Author

bretambrose
Feb 5, 2024
Maintainer

nabelekt Feb 5, 2024
Author

bretambrose Feb 5, 2024
Maintainer