Plugging out Ethernet cable crash library; if using TLS

[Meisterian]

Hello,

if using TLS; there is an assert that crash the library if the Ethernet cable is plugged out between the client and broker. This is the output from client_cli:

Debug build:

cli> underlying handshake:Success
v5::connack{rc:success,sp:false,ps:[{id:topic_alias_maximum,val:10},{id:assigned_client_identifier,val:auto-2A12395A-3258-96AA-4118-E078589C9A01},{id:receive_maximum,val:20}]}
recv error:stream truncated
client_cli: /tmp/async_mqtt/include/async_mqtt/util/impl/stream_close.hpp:61: void async_mqtt::detail::stream_impl<NextLayer>::stream_close_op::operator()(Self&, async_mqtt::error_code, std::reference_wrapper<Layer>) [with Self = boost::asio::detail::composed_op<async_mqtt::detail::stream_impl<boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::stream_close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::recv_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, network_manager<async_mqtt::basic_endpoint<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > > >, void(boost::system::error_code, async_mqtt::basic_packet_variant<2>)>, void()>, void(boost::system::error_code)>; Layer = boost::asio::basic_stream_socket<boost::asio::ip::tcp>; NextLayer = boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> >; async_mqtt::error_code = boost::system::error_code]: Assertion `state == close' failed.
Aborted

Release build:

cli> underlying handshake:Success
v5::connack{rc:success,sp:false,ps:[{id:topic_alias_maximum,val:10},{id:assigned_client_identifier,val:auto-6D779FFD-0B30-5EFF-841C-784F24CB5082},{id:receive_maximum,val:20}]}
recv error:stream truncated
Segmentation fault

, using a debugger shows that the state is "complete" and therefor the program crash. Doing the same test using TCP works without problem.

For now we have created a patch that replace the assert with an if statement. Then the library doesn't crash in either Debug or Release builds. This fix may of course have other impacts that we haven't seen yet.

Regards, Christian

[redboltz]

Thank you for reporting the issue.

I couldn't reproduce the issue on my environment. But my environment something different from yours because it is difficult to prepare plugging out environment.

I use three terminals.

Order	Role	Command
1	unplug emulation	socat tcp-listen:18883 tcp-connect:127.0.0.1:8883
2	broker	./broker --verbose 5
3	client	./client_cli --host 127.0.0.1 --port 18883 --protocol mqtts --verbose 5

Concept

client_cli -----> 18883 socat ------> 8883 broker

socat output

socat tcp-listen:18883 tcp-connect:127.0.0.1:8883

broker output

Set options:
auth_file                    : auth.json
bulk_read_buf_size           : 0
bulk_write                   : false
certificate                  : server.crt.pem
cfg                          : broker.conf
fixed_core_map               : false
iocs                         : 0
private_key                  : server.key.pem
recycling_allocator          : false
silent                       : false
tcp.port                     : 1883
tcp_no_delay                 : true
threads_per_ioc              : 0
tls.port                     : 8883
verbose                      : 5
verify_field                 : CN
verify_file                  : cacert.pem
ws.port                      : 10080
wss.port                     : 10443
18:07:50.223466 T:0x00007a63ba349840 S:info    C:mqtt_broker broker.cpp:157 iocs set to auto decide (0). Automatically set to 8
18:07:50.223616 T:0x00007a63ba349840 S:info    C:mqtt_broker broker.cpp:170 threads_per_ioc set to auto decide (0). Automatically set to 4
18:07:50.223678 T:0x00007a63ba349840 S:info    C:mqtt_broker broker.cpp:175 iocs:8 threads_per_ioc:4 total threads:32
18:07:50.223732 T:0x00007a63ba349840 S:info    C:mqtt_broker broker.cpp:185 auth_file:auth.json

client_cli output

Set options:
  cfg              : cli.conf
  clean_start      : true
  host             : 127.0.0.1
  keep_alive       : 0
  mqtt_version     : v5
  port             : 18883
  protocol         : mqtts
  sei              : 0
  verbose          : 5
  ws_path          : /
type 'help' to show menu
cli> underlying handshake:Success
v5::connack{rc:success,sp:false,ps:[{id:assigned_client_identifier,val:df564d30-942d-4733-9d47-9f1bcfe69fa2},{id:topic_alias_maximum,val:65535},{id:receive_maximum,val:65535}]}

---

When I kill socat, then I got the following result.

broker

18:12:23.645158 T:0x000070fb95c006c0 S:info    C:mqtt_impl endpoint_recv.hpp:37 A:0x5c1744887970 recv error:stream truncated
18:12:23.645234 T:0x000070fb95c006c0 S:info    C:mqtt_api endpoint_close.hpp:92 A:0x5c1744887970 close
18:12:23.645321 T:0x000070fb95c006c0 S:trace   C:mqtt_impl endpoint_close.hpp:41 A:0x5c1744887970 close initiate status:1
18:12:23.645549 T:0x000070fb95c006c0 S:info    C:mqtt_impl customized_ssl_stream.hpp:101 TLS async_shutdown ec:stream truncated
18:12:23.645620 T:0x000070fb95c006c0 S:info    C:mqtt_impl customized_basic_stream.hpp:45 TCP close
18:12:23.645691 T:0x000070fb95c006c0 S:trace   C:mqtt_impl endpoint_close.hpp:65 A:0x5c1744887970 close complete status:3
18:12:23.645743 T:0x000070fb95c006c0 S:trace   C:mqtt_impl endpoint_close.hpp:72 A:0x5c1744887970 process enqueued close
18:12:23.645804 T:0x000070fb95c006c0 S:info    C:mqtt_impl endpoint_recv.hpp:657 A:0x5c1744887970 recv code triggers close:stream truncated
18:12:23.645858 T:0x000070fb95c006c0 S:info    C:mqtt_broker broker.hpp:61 A:0x5c17448860c0 stream truncated
18:12:23.645918 T:0x000070fb95c006c0 S:trace   C:mqtt_broker broker.hpp:2095 A:0x5c17448860c0 close_proc
18:12:23.645975 T:0x000070fb95c006c0 S:trace   C:mqtt_broker broker.hpp:1931 A:0x5c17448860c0 close_proc_no_lock
18:12:23.646043 T:0x000070fb95c006c0 S:trace   C:mqtt_broker broker.hpp:2005 A:0x5c17448860c0 close cid:c106f643-926f-47ec-9017-62f65050d56f
18:12:23.646120 T:0x000070fb95c006c0 S:info    C:mqtt_api endpoint_close.hpp:119 A:0x5c17448860c0 close
18:12:23.646195 T:0x000070fb95c006c0 S:trace   C:mqtt_impl endpoint_close.hpp:59 A:0x5c1744887970 already closed
18:12:23.646245 T:0x000070fb95c006c0 S:info    C:mqtt_broker broker.hpp:2027 A:0x5c17448860c0 disconnect(optional)_and_closed
18:12:23.646326 T:0x000070fb95c006c0 S:trace   C:mqtt_impl endpoint_impl.ipp:473 A:0x5c17448860c0 destroy
18:12:23.646384 T:0x000070fb95c006c0 S:trace   C:mqtt_broker session_state.hpp:125 A:0x70fb6c0047f0 session destroy
18:12:23.646433 T:0x000070fb95c006c0 S:trace   C:mqtt_broker session_state.hpp:357 A:0x70fb6c0047f0 clean
18:12:23.646511 T:0x000070fb95c006c0 S:trace   C:mqtt_impl stream.hpp:59 A:0x5c1744887978 destroy
18:12:23.646574 T:0x000070fb95c006c0 S:trace   C:mqtt_impl stream_impl.hpp:64 A:0x5c1744887190 destroy

client_cli

recv error:stream truncated

---

Which version are you using? If you using async_mqtt from github clone, commit hash is helpful. And could you test the latest 0b1224641ad0cec389ff1053f7e8766e0a287e84 ?

Anyway, I will post my code analysis comment later.

[redboltz]

I added log outputting code to the following part:

https://github.com/redboltz/async_mqtt/blob/0b1224641ad0cec389ff1053f7e8766e0a287e84/include/async_mqtt/util/impl/stream_close.hpp#L29-L100

This is the log added code:

    template <typename Self>
    void operator()(
        Self& self,
        error_code ec = error_code{}
    ) {
        auto& a_strm{*strm};
        if (state == dispatch) {
            ASYNC_MQTT_LOG("mqtt_impl", fatal)
                << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                << "async operation start. state: dispatch -> close";
            state = close;
            as::dispatch(
                a_strm.get_executor(),
                as::append(
                    force_move(self),
                    error_code{},
                    std::ref(a_strm.nl_)
                )
            );
        }
        else {
            ASYNC_MQTT_LOG("mqtt_impl", fatal)
                << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                << "async operation finish. state: complete";
            BOOST_ASSERT(state == complete);
            a_strm.storing_cbs_.clear();
            a_strm.sending_cbs_.clear();
            self.complete(ec);
        }
    }

    template <typename Self, typename Layer>
    void operator()(
        Self& self,
        error_code /* ec */,
        std::reference_wrapper<Layer> stream
    ) {
        auto& a_strm{*strm};
        BOOST_ASSERT(state == close);
        if constexpr(has_async_close<Layer>::value) {
            ASYNC_MQTT_LOG("mqtt_impl", fatal)
                << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                << "has_async_close";
            if constexpr (has_next_layer<Layer>::value) {
                ASYNC_MQTT_LOG("mqtt_impl", fatal)
                    << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                    << "call async_close";
                layer_customize<Layer>::async_close(
                    stream.get(),
                    as::append(
                        force_move(self),
                        std::ref(stream.get().next_layer())
                    )
                );
            }
            else {
                ASYNC_MQTT_LOG("mqtt_impl", fatal)
                    << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                    << "NOT has_next_layer (lowest layer (TCP)). call async_close. "
                    << "state: close -> complete";
                state = complete;
                layer_customize<Layer>::async_close(
                    stream.get(),
                    force_move(self)
                );
            }
        }
        else {
            ASYNC_MQTT_LOG("mqtt_impl", fatal)
                << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                << "NOT has_async_close";
            if constexpr (has_next_layer<Layer>::value) {
                ASYNC_MQTT_LOG("mqtt_impl", fatal)
                    << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                    << "skip this layer";
                as::dispatch(
                    a_strm.get_executor(),
                    as::append(
                        force_move(self),
                        error_code{},
                        std::ref(a_strm.nl_)
                    )
                );
            }
            else {
                ASYNC_MQTT_LOG("mqtt_impl", fatal)
                    << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                    << "NOT has_next_layer (lowest layer (TCP)) state: close -> complete";
                state = complete;
                as::dispatch(
                    a_strm.get_executor(),
                    force_move(self)
                );
            }
        }
    }

Then I got the following result by the same operation (socat kill)

18:35:30.509316 T:0x000073b0f78ed840 S:fatal   C:mqtt_impl stream_close.hpp:36 A:0x55a024577260 async operation start. state: dispatch -> close
18:35:30.509489 T:0x000073b0f78ed840 S:fatal   C:mqtt_impl stream_close.hpp:69 A:0x55a024577260 has_async_close
18:35:30.509548 T:0x000073b0f78ed840 S:fatal   C:mqtt_impl stream_close.hpp:73 A:0x55a024577260 call async_close
18:35:30.509689 T:0x000073b0f78ed840 S:fatal   C:mqtt_impl stream_close.hpp:69 A:0x55a024577260 has_async_close
18:35:30.509725 T:0x000073b0f78ed840 S:fatal   C:mqtt_impl stream_close.hpp:85 A:0x55a024577260 NOT has_next_layer (lowest layer (TCP)). call async_close. state: close -> complete
18:35:30.509784 T:0x000073b0f78ed840 S:fatal   C:mqtt_impl stream_close.hpp:50 A:0x55a024577260 async operation finish. state: complete
recv error:stream truncated

This is the expected log outputs.

If you replaced this part of the code with my log added version and cheking the log, it is very helpful to analyze the issue. Could you try this?

[Meisterian]

We checkout the 9.0.0 tag. I can only reproduce this by plugging out the cable. For example killing the broker (mosquitto) with kill -9, will give the same output as you showed. Here is the output you requested:

cli> underlying handshake:Success
v5::connack{rc:success,sp:false,ps:[{id:topic_alias_maximum,val:10},{id:assigned_client_identifier,val:auto-7B686406-6719-A6ED-F768-474A2223E171},{id:receive_maximum,val:20}]}
13:05:13.181037 T:0x00007f4217cba480 S:fatal   C:mqtt_impl stream_close.hpp:28 A:0x1f2a980 async operation start. state: dispatch -> close
13:05:13.181253 T:0x00007f4217cba480 S:fatal   C:mqtt_impl stream_close.hpp:52 A:0x1f2a980 has_async_close
13:05:13.181300 T:0x00007f4217cba480 S:fatal   C:mqtt_impl stream_close.hpp:55 A:0x1f2a980 call async_close
13:05:16.181962 T:0x00007f4217cba480 S:fatal   C:mqtt_impl stream_close.hpp:52 A:0x1f2a980 has_async_close
13:05:16.182212 T:0x00007f4217cba480 S:fatal   C:mqtt_impl stream_close.hpp:61 A:0x1f2a980 NOT has_next_layer (lowest layer (TCP)). call async_close. state: close -> complete
13:05:16.182447 T:0x00007f4217cba480 S:fatal   C:mqtt_impl stream_close.hpp:36 A:0x1f2a980 async operation finish. state: complete
recv error:stream truncated
client_cli: /tmp/async_mqtt/include/async_mqtt/util/impl/stream_close.hpp:50: void async_mqtt::detail::stream_impl<NextLayer>::stream_close_op::operator()(Self&, async_mqtt::error_code, std::reference_wrapper<Layer>) [with Self = boost::asio::detail::composed_op<async_mqtt::detail::stream_impl<boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::stream_close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::recv_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, network_manager<async_mqtt::basic_endpoint<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > > >, void(boost::system::error_code, async_mqtt::basic_packet_variant<2>)>, void()>, void(boost::system::error_code)>; Layer = boost::asio::basic_stream_socket<boost::asio::ip::tcp>; NextLayer = boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> >; async_mqtt::error_code = boost::system::error_code]: Assertion `state == close' failed.
Aborted

[Meisterian]

And this is from main(0b1224641ad0cec389ff1053f7e8766e0a287e84):

cli> underlying handshake:Success
v5::connack{rc:success,sp:false,ps:[{id:topic_alias_maximum,val:10},{id:assigned_client_identifier,val:auto-3D00D7B0-BFDD-1536-613C-AF238C0C7804},{id:receive_maximum,val:20}]}
13:39:16.944066 T:0x00007f4838c8f480 S:fatal   C:mqtt_impl stream_close.hpp:28 A:0x1da2980 async operation start. state: dispatch -> close
13:39:16.944229 T:0x00007f4838c8f480 S:fatal   C:mqtt_impl stream_close.hpp:52 A:0x1da2980 has_async_close
13:39:16.944261 T:0x00007f4838c8f480 S:fatal   C:mqtt_impl stream_close.hpp:55 A:0x1da2980 call async_close
13:39:19.945202 T:0x00007f4838c8f480 S:fatal   C:mqtt_impl stream_close.hpp:52 A:0x1da2980 has_async_close
13:39:19.945450 T:0x00007f4838c8f480 S:fatal   C:mqtt_impl stream_close.hpp:61 A:0x1da2980 NOT has_next_layer (lowest layer (TCP)). call async_close. state: close -> complete
13:39:19.945684 T:0x00007f4838c8f480 S:fatal   C:mqtt_impl stream_close.hpp:36 A:0x1da2980 async operation finish. state: complete
recv error:stream truncated
client_cli: /tmp/async_mqtt/include/async_mqtt/util/impl/stream_close.hpp:50: void async_mqtt::detail::stream_impl<NextLayer>::stream_close_op::operator()(Self&, async_mqtt::error_code, std::reference_wrapper<Layer>) [with Self = boost::asio::detail::composed_op<async_mqtt::detail::stream_impl<boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::stream_close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::recv_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, network_manager<async_mqtt::basic_endpoint<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > > >, void(boost::system::error_code, async_mqtt::basic_packet_variant<2>)>, void()>, void(boost::system::error_code)>; Layer = boost::asio::basic_stream_socket<boost::asio::ip::tcp>; NextLayer = boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> >; async_mqtt::error_code = boost::system::error_code]: Assertion `state == close' failed.
Aborted

[redboltz]

tag 9.0.0 os https://github.com/redboltz/async_mqtt/blob/9.0.0/include/async_mqtt/util/impl/stream_close.hpp

It seems that the line number is something weird. The first log is line 28. The assertion failed is line 50. But it doesn't match 9.0.0. Could you check your code?

Here is 9.0.0 with log code from the line 1.

// Copyright Takatoshi Kondo 2022
//
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)

#if !defined(ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP)
#define ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP

#include <async_mqtt/util/stream.hpp>
#include <async_mqtt/protocol_version.hpp>

namespace async_mqtt {

namespace detail {

template <typename NextLayer>
struct stream_impl<NextLayer>::stream_close_op {
    using stream_type = this_type;
    using stream_type_sp = std::shared_ptr<stream_type>;

    std::shared_ptr<stream_type> strm;
    enum {
        dispatch,
        close,
        complete
    } state = dispatch;

    template <typename Self>
    void operator()(
        Self& self,
        error_code ec = error_code{}
    ) {
        auto& a_strm{*strm};
        if (state == dispatch) {
            ASYNC_MQTT_LOG("mqtt_impl", fatal)
                << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                << "async operation start. state: dispatch -> close";
            state = close;
            as::dispatch(
                a_strm.get_executor(),
                as::append(
                    force_move(self),
                    error_code{},
                    std::ref(a_strm.nl_)
                )
            );
        }
        else {
            ASYNC_MQTT_LOG("mqtt_impl", fatal)
                << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                << "async operation finish. state: complete";
            BOOST_ASSERT(state == complete);
            a_strm.storing_cbs_.clear();
            a_strm.sending_cbs_.clear();
            self.complete(ec);
        }
    }

    template <typename Self, typename Layer>
    void operator()(
        Self& self,
        error_code /* ec */,
        std::reference_wrapper<Layer> stream
    ) {
        auto& a_strm{*strm};
        BOOST_ASSERT(state == close);
        if constexpr(has_async_close<Layer>::value) {
            ASYNC_MQTT_LOG("mqtt_impl", fatal)
                << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                << "has_async_close";
            if constexpr (has_next_layer<Layer>::value) {
                ASYNC_MQTT_LOG("mqtt_impl", fatal)
                    << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                    << "call async_close";
                layer_customize<Layer>::async_close(
                    stream.get(),
                    as::append(
                        force_move(self),
                        std::ref(stream.get().next_layer())
                    )
                );
            }
            else {
                ASYNC_MQTT_LOG("mqtt_impl", fatal)
                    << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                    << "NOT has_next_layer (lowest layer (TCP)). call async_close. "
                    << "state: close -> complete";
                state = complete;
                layer_customize<Layer>::async_close(
                    stream.get(),
                    force_move(self)
                );
            }
        }
        else {
            ASYNC_MQTT_LOG("mqtt_impl", fatal)
                << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                << "NOT has_async_close";
            if constexpr (has_next_layer<Layer>::value) {
                ASYNC_MQTT_LOG("mqtt_impl", fatal)
                    << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                    << "skip this layer";
                as::dispatch(
                    a_strm.get_executor(),
                    as::append(
                        force_move(self),
                        error_code{},
                        std::ref(a_strm.nl_)
                    )
                );
            }
            else {
                ASYNC_MQTT_LOG("mqtt_impl", fatal)
                    << ASYNC_MQTT_ADD_VALUE(address, strm.get())
                    << "NOT has_next_layer (lowest layer (TCP)) state: close -> complete";
                state = complete;
                as::dispatch(
                    a_strm.get_executor(),
                    force_move(self)
                );
            }
        }
    }
};

[redboltz]

And this is from main(0b12246):

cli> underlying handshake:Success v5::connack{rc:success,sp:false,ps:[{id:topic_alias_maximum,val:10},{id:assigned_client_identifier,val:auto-3D00D7B0-BFDD-1536-613C-AF238C0C7804},{id:receive_maximum,val:20}]} 13:39:16.944066 T:0x00007f4838c8f480 S:fatal C:mqtt_impl stream_close.hpp:28 A:0x1da2980 async operation start. state: dispatch -> close 13:39:16.944229 T:0x00007f4838c8f480 S:fatal C:mqtt_impl stream_close.hpp:52 A:0x1da2980 has_async_close 13:39:16.944261 T:0x00007f4838c8f480 S:fatal C:mqtt_impl stream_close.hpp:55 A:0x1da2980 call async_close 13:39:19.945202 T:0x00007f4838c8f480 S:fatal C:mqtt_impl stream_close.hpp:52 A:0x1da2980 has_async_close 13:39:19.945450 T:0x00007f4838c8f480 S:fatal C:mqtt_impl stream_close.hpp:61 A:0x1da2980 NOT has_next_layer (lowest layer (TCP)). call async_close. state: close -> complete 13:39:19.945684 T:0x00007f4838c8f480 S:fatal C:mqtt_impl stream_close.hpp:36 A:0x1da2980 async operation finish. state: complete recv error:stream truncated client_cli: /tmp/async_mqtt/include/async_mqtt/util/impl/stream_close.hpp:50: void async_mqtt::detail::stream_impl::stream_close_op::operator()(Self&, async_mqtt::error_code, std::reference_wrapper) [with Self = boost::asio::detail::composed_op<async_mqtt::detail::stream_impl<boost::asio::ssl::stream >::stream_close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream >::close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream >::recv_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, network_manager<async_mqtt::basic_endpoint<async_mqtt::role::client, 2, boost::asio::ssl::stream > >, void(boost::system::error_code, async_mqtt::basic_packet_variant<2>)>, void()>, void(boost::system::error_code)>; Layer = boost::asio::basic_stream_socketboost::asio::ip::tcp; NextLayer = boost::asio::ssl::stream; async_mqtt::error_code = boost::system::error_code]: Assertion `state == close' failed. Aborted

https://github.com/redboltz/async_mqtt/blob/0b1224641ad0cec389ff1053f7e8766e0a287e84/include/async_mqtt/util/impl/stream_close.hpp#L28 is a blank line. No logs added before the line 28. It is something weird too.

[Meisterian]

Sorry, the problem is that I have automatic code formatting.

// Copyright Takatoshi Kondo 2022
//
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)

#if !defined(ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP)
#define ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP

#include <async_mqtt/protocol_version.hpp>
#include <async_mqtt/util/stream.hpp>

namespace async_mqtt {

namespace detail {

template <typename NextLayer> struct stream_impl<NextLayer>::stream_close_op {
  using stream_type = this_type;
  using stream_type_sp = std::shared_ptr<stream_type>;

  std::shared_ptr<stream_type> strm;
  enum { dispatch, close, complete } state = dispatch;

  template <typename Self>
  void operator()(Self &self, error_code ec = error_code{}) {
    auto &a_strm{*strm};
    if (state == dispatch) {
      ASYNC_MQTT_LOG("mqtt_impl", fatal)
          << ASYNC_MQTT_ADD_VALUE(address, strm.get())
          << "async operation start. state: dispatch -> close";
      state = close;
      as::dispatch(
          a_strm.get_executor(),
          as::append(force_move(self), error_code{}, std::ref(a_strm.nl_)));
    } else {
      ASYNC_MQTT_LOG("mqtt_impl", fatal)
          << ASYNC_MQTT_ADD_VALUE(address, strm.get())
          << "async operation finish. state: complete";
      BOOST_ASSERT(state == complete);
      a_strm.storing_cbs_.clear();
      a_strm.sending_cbs_.clear();
      self.complete(ec);
    }
  }

  template <typename Self, typename Layer>
  void operator()(Self &self, error_code /* ec */,
                  std::reference_wrapper<Layer> stream) {
    auto &a_strm{*strm};
    BOOST_ASSERT(state == close);
    if constexpr (has_async_close<Layer>::value) {
      ASYNC_MQTT_LOG("mqtt_impl", fatal)
          << ASYNC_MQTT_ADD_VALUE(address, strm.get()) << "has_async_close";
      if constexpr (has_next_layer<Layer>::value) {
        ASYNC_MQTT_LOG("mqtt_impl", fatal)
            << ASYNC_MQTT_ADD_VALUE(address, strm.get()) << "call async_close";
        layer_customize<Layer>::async_close(
            stream.get(),
            as::append(force_move(self), std::ref(stream.get().next_layer())));
      } else {
        ASYNC_MQTT_LOG("mqtt_impl", fatal)
            << ASYNC_MQTT_ADD_VALUE(address, strm.get())
            << "NOT has_next_layer (lowest layer (TCP)). call async_close. "
            << "state: close -> complete";
        state = complete;
        layer_customize<Layer>::async_close(stream.get(), force_move(self));
      }
    } else {
      ASYNC_MQTT_LOG("mqtt_impl", fatal)
          << ASYNC_MQTT_ADD_VALUE(address, strm.get()) << "NOT has_async_close";
      if constexpr (has_next_layer<Layer>::value) {
        ASYNC_MQTT_LOG("mqtt_impl", fatal)
            << ASYNC_MQTT_ADD_VALUE(address, strm.get()) << "skip this layer";
        as::dispatch(
            a_strm.get_executor(),
            as::append(force_move(self), error_code{}, std::ref(a_strm.nl_)));
      } else {
        ASYNC_MQTT_LOG("mqtt_impl", fatal)
            << ASYNC_MQTT_ADD_VALUE(address, strm.get())
            << "NOT has_next_layer (lowest layer (TCP)) state: close -> "
               "complete";
        state = complete;
        as::dispatch(a_strm.get_executor(), force_move(self));
      }
    }
  }
};
} // namespace detail

template <typename NextLayer>
template <typename CompletionToken>
BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(CompletionToken, void())
stream<NextLayer>::async_close(CompletionToken &&token) {
  BOOST_ASSERT(impl_);
  return as::async_compose<CompletionToken, void(error_code)>(
      typename impl_type::stream_close_op{impl_}, token, get_executor());
}

} // namespace async_mqtt

#endif // ASYNC_MQTT_UTIL_IMPL_STREAM_CLOSE_HPP`

[redboltz]

Please use

```cpp your code ```

notation.

[redboltz]

Please keep the original code. Auto code fomatting confused me.

[redboltz]

I edited your comment only for applying ```. I will prepare actual (phisical) environment to reproduce the issue. Please wait a moment.

It seems that the async close sequence itself works expectedly. However, 2nd overload of operator() is called after the sequence is completed. I don't know why, so far.

[Meisterian]

Okay, I changed the file using vim instead. Posting the output that should match your numbering.

cli> underlying handshake:Success
v5::connack{rc:success,sp:false,ps:[{id:topic_alias_maximum,val:10},{id:assigned_client_identifier,val:auto-6085DB93-EC69-0F3E-BF77-EDBAA2233E92},{id:receive_maximum,val:20}]}
14:00:22.486021 T:0x00007fcda27d5480 S:fatal   C:mqtt_impl stream_close.hpp:36 A:0x227a980 async operation start. state: dispatch -> close
14:00:22.486327 T:0x00007fcda27d5480 S:fatal   C:mqtt_impl stream_close.hpp:69 A:0x227a980 has_async_close
14:00:22.486443 T:0x00007fcda27d5480 S:fatal   C:mqtt_impl stream_close.hpp:73 A:0x227a980 call async_close
14:00:25.486829 T:0x00007fcda27d5480 S:fatal   C:mqtt_impl stream_close.hpp:69 A:0x227a980 has_async_close
14:00:25.487027 T:0x00007fcda27d5480 S:fatal   C:mqtt_impl stream_close.hpp:85 A:0x227a980 NOT has_next_layer (lowest layer (TCP)). call async_close. state: close -> complete
14:00:25.487222 T:0x00007fcda27d5480 S:fatal   C:mqtt_impl stream_close.hpp:50 A:0x227a980 async operation finish. state: complete
recv error:stream truncated
client_cli: /tmp/async_mqtt/include/async_mqtt/util/impl/stream_close.hpp:67: void async_mqtt::detail::stream_impl<NextLayer>::stream_close_op::operator()(Self&, async_mqtt::error_code, std::reference_wrapper<Layer>) [with Self = boost::asio::detail::composed_op<async_mqtt::detail::stream_impl<boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::stream_close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::recv_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, network_manager<async_mqtt::basic_endpoint<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > > >, void(boost::system::error_code, async_mqtt::basic_packet_variant<2>)>, void()>, void(boost::system::error_code)>; Layer = boost::asio::basic_stream_socket<boost::asio::ip::tcp>; NextLayer = boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> >; async_mqtt::error_code = boost::system::error_code]: Assertion `state == close' failed.
Aborted

[redboltz]

By the way, I prepared trial MQTT broker that support TCP, TLS, ws, and wss. You can connect to the broker as follows:

./client_cli --host mqtt.redboltz.net --port 8883 --protocol mqtts --verbose 5

When you unplugged your client machine, what is happened?

On my environment, no disconnection is happened (I wait 1 minutes). I guess that It depends on TCP setting.

[redboltz]

You mentioned that you used degger, and at the line 67, assertion failed line, you checked the variable state and its value is complete. It is matched to the log output. Could you print a back trace before the assertion fail ? NOTE: line 67 passed 5 times. The last one is problematic, and I need the backtrace of the 5th passing.

I think that it helps to know where the unexpected call is come from.

Thank you very much for your cooperation.

[redboltz]

By the way which version of Boost are you using ?

[Meisterian]

I'm using 1.84. I will drive to my company now and see if I can try to connect to your server using an ethernet cable.

[redboltz]

It is not the solution of this issue but I created #354 that fixes loglevel setting issue of client_cli. client_cli didn't output trace level log. I will merge the #354 if CI is passed. It helps this issue analysis.

[redboltz]

354 has been merged. Now, client_cli log works fine.

[redboltz]

354 has been merged. Now, client_cli log works fine.

Please re-run using updated main branch. We can get more detailed logs. It would help to understand why the handler calls unexpectedly.

[redboltz]

I create #355,. I am not 100% sure but It would fix this issue. TLS with plug out seems to relate to async_shutdown() and its timeout.

355 re-designs timeout logic.

Please try it too. (#355 doesn't include https://github.com/redboltz/async_mqtt/issues/353#issuecomment-2382955719 logs)

355 with the log patch is the best information.

[Meisterian]

Here is the output. I had to set keep-alive in order to crash it using your server:

./client_cli --host mqtt.redboltz.net --port 8883 --protocol mqtts --verbose 5 --keep_alive 5
Set options:
  cfg              : cli.conf
  clean_start      : true
  host             : mqtt.redboltz.net
  keep_alive       : 5
  mqtt_version     : v5
  port             : 8883
  protocol         : mqtts
  sei              : 0
  verbose          : 5
  ws_path          : /
type 'help' to show menu
cli> 16:23:26.609184 T:0x00007f3af7a2c480 S:info    C:mqtt_api endpoint_impl.ipp:482 A:0x7ffe83bc02d0 set_auto_pub_response val:true
underlying handshake:Success
16:23:27.944043 T:0x00007f3af7a2c480 S:info    C:mqtt_api endpoint_send.hpp:664 A:0x7ffe83bc02d0 send:v5::connect{cid:,ka:5,cs:true}
16:23:27.944212 T:0x00007f3af7a2c480 S:info    C:mqtt_impl store.hpp:98 [store] clear
16:23:27.944540 T:0x00007f3af7a2c480 S:info    C:mqtt_api endpoint_recv.hpp:692 A:0x7ffe83bc02d0 recv
16:23:28.205654 T:0x00007f3af7a2c480 S:trace   C:mqtt_impl endpoint_recv.hpp:134 A:0x1b776c0 recv:v5::connack{rc:success,sp:false,ps:[{id:assigned_client_identifier,val:5c4fec0d-f633-4169-bef1-edfef703b178},{id:topic_alias_maximum,val:65535},{id:receive_maximum,val:65535}]}
16:23:28.205909 T:0x00007f3af7a2c480 S:info    C:mqtt_impl store.hpp:98 [store] clear
v5::connack{rc:success,sp:false,ps:[{id:assigned_client_identifier,val:5c4fec0d-f633-4169-bef1-edfef703b178},{id:topic_alias_maximum,val:65535},{id:receive_maximum,val:65535}]}
16:23:28.206048 T:0x00007f3af7a2c480 S:info    C:mqtt_api endpoint_recv.hpp:692 A:0x7ffe83bc02d0 recv
16:24:03.654018 T:0x00007f3af7a2c480 S:info    C:mqtt_impl endpoint_recv.hpp:37 A:0x1b776c0 recv error:stream truncated
16:24:03.654189 T:0x00007f3af7a2c480 S:info    C:mqtt_api endpoint_close.hpp:92 A:0x1b776c0 close
16:24:03.654286 T:0x00007f3af7a2c480 S:trace   C:mqtt_impl endpoint_close.hpp:41 A:0x1b776c0 close initiate status:1
16:24:03.654346 T:0x00007f3af7a2c480 S:fatal   C:mqtt_impl stream_close.hpp:36 A:0x1b77c50 async operation start. state: dispatch -> close
16:24:03.654404 T:0x00007f3af7a2c480 S:fatal   C:mqtt_impl stream_close.hpp:69 A:0x1b77c50 has_async_close
16:24:03.654443 T:0x00007f3af7a2c480 S:fatal   C:mqtt_impl stream_close.hpp:73 A:0x1b77c50 call async_close
16:24:06.655051 T:0x00007f3af7a2c480 S:info    C:mqtt_impl customized_ssl_stream.hpp:93 TLS async_shutdown timeout
16:24:06.655281 T:0x00007f3af7a2c480 S:fatal   C:mqtt_impl stream_close.hpp:69 A:0x1b77c50 has_async_close
16:24:06.655395 T:0x00007f3af7a2c480 S:fatal   C:mqtt_impl stream_close.hpp:85 A:0x1b77c50 NOT has_next_layer (lowest layer (TCP)). call async_close. state: close -> complete
16:24:06.655516 T:0x00007f3af7a2c480 S:info    C:mqtt_impl customized_basic_stream.hpp:50 TCP close
16:24:06.655763 T:0x00007f3af7a2c480 S:fatal   C:mqtt_impl stream_close.hpp:50 A:0x1b77c50 async operation finish. state: complete
16:24:06.655889 T:0x00007f3af7a2c480 S:trace   C:mqtt_impl endpoint_close.hpp:65 A:0x1b776c0 close complete status:3
16:24:06.656033 T:0x00007f3af7a2c480 S:trace   C:mqtt_impl endpoint_close.hpp:72 A:0x1b776c0 process enqueued close
16:24:06.656134 T:0x00007f3af7a2c480 S:info    C:mqtt_impl endpoint_recv.hpp:657 A:0x1b776c0 recv code triggers close:stream truncated
recv error:stream truncated
16:24:06.656254 T:0x00007f3af7a2c480 S:info    C:mqtt_impl customized_ssl_stream.hpp:105 TLS async_shutdown ec:Operation canceled
client_cli: /tmp/async_mqtt/include/async_mqtt/util/impl/stream_close.hpp:67: void async_mqtt::detail::stream_impl<NextLayer>::stream_close_op::operator()(Self&, async_mqtt::error_code, std::reference_wrapper<Layer>) [with Self = boost::asio::detail::composed_op<async_mqtt::detail::stream_impl<boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::stream_close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::close_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, boost::asio::detail::composed_op<async_mqtt::detail::basic_endpoint_impl<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > >::recv_op, boost::asio::detail::composed_work<void(boost::asio::any_io_executor)>, network_manager<async_mqtt::basic_endpoint<async_mqtt::role::client, 2, boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> > > >, void(boost::system::error_code, async_mqtt::basic_packet_variant<2>)>, void()>, void(boost::system::error_code)>; Layer = boost::asio::basic_stream_socket<boost::asio::ip::tcp>; NextLayer = boost::asio::ssl::stream<boost::asio::basic_stream_socket<boost::asio::ip::tcp> >; async_mqtt::error_code = boost::system::error_code]: Assertion `state == close' failed.
Aborted

[redboltz]

Thank you for reporitng.

I think that the following line is the reason of the problematic behavior.

16:24:06.656254 T:0x00007f3af7a2c480 S:info    C:mqtt_impl customized_ssl_stream.hpp:105 TLS async_shutdown ec:Operation canceled

If my hypothetis is right, the issue should be solved by #355. Please try it.

[Meisterian]

To your server, it seems to work! I will test my original setup as well, but here is the output connecting to your server:

./client_cli --host mqtt.redboltz.net --port 8883 --protocol mqtts --verbose 5 --keep_alive 5
Set options:
  cfg              : cli.conf
  clean_start      : true
  host             : mqtt.redboltz.net
  keep_alive       : 5
  mqtt_version     : v5
  port             : 8883
  protocol         : mqtts
  sei              : 0
  verbose          : 5
  ws_path          : /
type 'help' to show menu
cli> 16:45:23.471228 T:0x00007f930c889480 S:info    C:mqtt_api endpoint_impl.ipp:482 A:0x7fffce752e00 set_auto_pub_response val:true
underlying handshake:Success
16:45:24.906210 T:0x00007f930c889480 S:info    C:mqtt_api endpoint_send.hpp:664 A:0x7fffce752e00 send:v5::connect{cid:,ka:5,cs:true}
16:45:24.906412 T:0x00007f930c889480 S:info    C:mqtt_impl store.hpp:98 [store] clear
16:45:24.906780 T:0x00007f930c889480 S:info    C:mqtt_api endpoint_recv.hpp:692 A:0x7fffce752e00 recv
16:45:25.179255 T:0x00007f930c889480 S:trace   C:mqtt_impl endpoint_recv.hpp:134 A:0x2f186c0 recv:v5::connack{rc:success,sp:false,ps:[{id:assigned_client_identifier,val:e294e310-0915-48fd-8a49-cce6b14ff42c},{id:topic_alias_maximum,val:65535},{id:receive_maximum,val:65535}]}
16:45:25.179460 T:0x00007f930c889480 S:info    C:mqtt_impl store.hpp:98 [store] clear
v5::connack{rc:success,sp:false,ps:[{id:assigned_client_identifier,val:e294e310-0915-48fd-8a49-cce6b14ff42c},{id:topic_alias_maximum,val:65535},{id:receive_maximum,val:65535}]}
16:45:25.179582 T:0x00007f930c889480 S:info    C:mqtt_api endpoint_recv.hpp:692 A:0x7fffce752e00 recv
16:45:30.158889 T:0x00007f930c889480 S:trace   C:mqtt_impl endpoint_recv.hpp:134 A:0x2f186c0 recv:v5::pingresp{}
v5::pingresp{}
16:45:30.159119 T:0x00007f930c889480 S:info    C:mqtt_api endpoint_recv.hpp:692 A:0x7fffce752e00 recv
16:45:35.157655 T:0x00007f930c889480 S:trace   C:mqtt_impl endpoint_recv.hpp:134 A:0x2f186c0 recv:v5::pingresp{}
v5::pingresp{}
16:45:35.157932 T:0x00007f930c889480 S:info    C:mqtt_api endpoint_recv.hpp:692 A:0x7fffce752e00 recv
16:46:04.636602 T:0x00007f930c889480 S:info    C:mqtt_impl endpoint_recv.hpp:37 A:0x2f186c0 recv error:stream truncated
16:46:04.636800 T:0x00007f930c889480 S:info    C:mqtt_api endpoint_close.hpp:92 A:0x2f186c0 close
16:46:04.636925 T:0x00007f930c889480 S:trace   C:mqtt_impl endpoint_close.hpp:41 A:0x2f186c0 close initiate status:1
16:46:04.637039 T:0x00007f930c889480 S:fatal   C:mqtt_impl stream_close.hpp:36 A:0x2f18c50 async operation start. state: dispatch -> close
16:46:04.637131 T:0x00007f930c889480 S:fatal   C:mqtt_impl stream_close.hpp:69 A:0x2f18c50 has_async_close
16:46:04.637215 T:0x00007f930c889480 S:fatal   C:mqtt_impl stream_close.hpp:73 A:0x2f18c50 call async_close
16:46:07.637930 T:0x00007f930c889480 S:info    C:mqtt_impl customized_ssl_stream.hpp:98 TLS async_shutdown timeout
16:46:07.638287 T:0x00007f930c889480 S:fatal   C:mqtt_impl stream_close.hpp:69 A:0x2f18c50 has_async_close
16:46:07.638397 T:0x00007f930c889480 S:fatal   C:mqtt_impl stream_close.hpp:85 A:0x2f18c50 NOT has_next_layer (lowest layer (TCP)). call async_close. state: close -> complete
16:46:07.638482 T:0x00007f930c889480 S:info    C:mqtt_impl customized_basic_stream.hpp:50 TCP close
16:46:07.638676 T:0x00007f930c889480 S:fatal   C:mqtt_impl stream_close.hpp:50 A:0x2f18c50 async operation finish. state: complete
16:46:07.638790 T:0x00007f930c889480 S:trace   C:mqtt_impl endpoint_close.hpp:65 A:0x2f186c0 close complete status:3
16:46:07.638865 T:0x00007f930c889480 S:trace   C:mqtt_impl endpoint_close.hpp:72 A:0x2f186c0 process enqueued close
16:46:07.638951 T:0x00007f930c889480 S:info    C:mqtt_impl endpoint_recv.hpp:657 A:0x2f186c0 recv code triggers close:stream truncated
recv error:stream truncated
16:46:07.639113 T:0x00007f930c889480 S:info    C:mqtt_impl customized_ssl_stream.hpp:116 TLS async_shutdown already timeout

[Meisterian]

I tested my original setup and it works as well! Thank you for the quick response!

[redboltz]

Finally I reproduced the issue.

Here is the sequence using one phisical machine.

Step	Action	Command
1	run socat	socat tcp-listen:18883 tcp-connect:127.0.0.1:8883
2	run broker	./broker --verbose 5
3	run client	./client_cli --host 127.0.0.1 --port 18883 --protocol mqtts --keep_alive 5 --verbose 5
4	kill broker	Press Ctrl-C at the broker running terminal

Important point is kill broker not socat. Making the situation that client_cli can't know disconnection directly.

I've tested as follows:

• main 080c6e15435dbc3ee210ea9a3e49c91e0e6853a5
  • The issue (assertion falied) is reproduced
• fix_353 branch 676a0b76ea8554ab53a142844921005da3b325c
  • The issue (assertion falied) is not reproduced (solved)

I will release the fix as 9.0.1 soon.

[redboltz]

355 has been merged, and version 9.0.1 has been released.

Thank you for all your hard work; it was truly appreciated and valuable.

[Meisterian]

No problem! Thank you for resolving the issue so quickly!