fix: serialize MQTT ws I/O and dewcouple keepalive

src/slic3r/Utils/OrcaMqttConnection.cpp

@@ -23,6 +23,12 @@ struct OrcaMqttConnection::Connection {
     boost::asio::io_context io_context;
     boost::asio::ssl::context ssl_context;
     boost::asio::ip::tcp::resolver resolver;
+    boost::asio::steady_timer keepalive_timer;
+    boost::beast::flat_buffer read_buffer;
+    std::deque<std::shared_ptr<std::vector<uint8_t>>> outbound_packets;
+    boost::system::error_code terminal_error;
+    std::atomic_bool async_session_started{false};
+    bool write_in_progress{false};
     // Exactly one of these is engaged once ws_handshake() has run: wss for
     // wss:// endpoints, ws for plaintext ws://.
     std::optional<TlsWebSocket>   wss;
@@ -31,6 +37,7 @@ struct OrcaMqttConnection::Connection {
     Connection()
         : ssl_context(boost::asio::ssl::context::tls_client)
         , resolver(io_context)
+        , keepalive_timer(io_context)
     {}
 };
 
@@ -81,13 +88,24 @@ void OrcaMqttConnection::stop() {
     {
         std::lock_guard<std::mutex> lock(connection_mutex);
         if (active_connection) {
-            // Generic so it accepts either the TLS or the plaintext websocket.
+            if (active_connection->async_session_started.load()) {
+                // The worker owns the live WebSocket. Stop dispatching its
+                // asynchronous operations; the worker closes the socket after
+                // leaving the event loop.
+                active_connection->io_context.stop();
+            } else {
+                // Setup still uses synchronous operations on the worker. Wake a
+                // pending resolve/connect/CONNACK read without competing with a
+                // live asynchronous session.
                 auto shutdown_socket = [](auto& websocket) {
                     auto& socket = boost::beast::get_lowest_layer(websocket).socket();
                     boost::system::error_code socket_error;
-                socket.cancel(socket_error);
+                    if (socket.cancel(socket_error))
-                socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, socket_error);
+                        return;
-                socket.close(socket_error);
+                    if (socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, socket_error))
+                        return;
+                    if (socket.close(socket_error))
+                        return;
                 };
                 if (active_connection->wss)
                     shutdown_socket(*active_connection->wss);
@@ -96,6 +114,7 @@ void OrcaMqttConnection::stop() {
                 active_connection->resolver.cancel();
             }
         }
+    }
     if (worker.joinable())
         worker.join();
 
@@ -130,13 +149,10 @@ void OrcaMqttConnection::flush_subscription_change() {
     }
     if (!conn || !connacked)
         return; // no live MQTT session yet — the worker sends the set on CONNACK
-
+    boost::asio::post(conn->io_context, [this, conn] {
-    // beast permits a concurrent writer while the worker is blocked in
+        if (!stopping.load() && connected.load())
-    // websocket.read(); every write is serialised by write_mutex inside ws_write().
+            send_pending_subscriptions(conn);
-    try {
+    });
-        send_pending_subscriptions(*conn);
-    } catch (const std::exception&) {
-    }
 }
 
 bool OrcaMqttConnection::subscribe(const std::string& dev_id) {
@@ -157,7 +173,7 @@ bool OrcaMqttConnection::subscribe(const std::string& dev_id) {
         pending_subscriptions.insert(topic);
     }
     state_cv.notify_all();
-    flush_subscription_change(); // emit SUBSCRIBE now on the live socket (no reconnect)
+    flush_subscription_change(); // ask the worker to emit SUBSCRIBE now (no reconnect)
     return true;
 }
 
@@ -183,7 +199,7 @@ bool OrcaMqttConnection::unsubscribe(const std::string& dev_id) {
         }
     }
     state_cv.notify_all();
-    flush_subscription_change(); // emit UNSUBSCRIBE now on the live socket (no reconnect)
+    flush_subscription_change(); // ask the worker to emit UNSUBSCRIBE now (no reconnect)
     return true;
 }
 
@@ -298,10 +314,6 @@ void OrcaMqttConnection::ws_write(Connection& conn, const std::vector<uint8_t>&
     if (packet.empty()) {
         return;
     }
-    // Writes come from the worker thread AND, for dynamic (un)subscribes, the
-    // caller thread. Serialise them; the worker's concurrent read is fine (beast
-    // allows one reader + one writer).
-    std::lock_guard<std::mutex> lock(write_mutex);
     if (conn.wss) {
         conn.wss->binary(true);
         conn.wss->write(boost::asio::buffer(packet));
@@ -311,6 +323,110 @@ void OrcaMqttConnection::ws_write(Connection& conn, const std::vector<uint8_t>&
     }
 }
 
+void OrcaMqttConnection::close_connection(Connection& conn) {
+    boost::system::error_code error;
+    if (conn.wss) {
+        auto& socket = boost::beast::get_lowest_layer(*conn.wss).socket();
+        if (socket.cancel(error))
+            return;
+        if (socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, error))
+            return;
+        if (socket.close(error))
+            return;
+    } else if (conn.ws) {
+        auto& socket = boost::beast::get_lowest_layer(*conn.ws).socket();
+        if (socket.cancel(error))
+            return;
+        if (socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, error))
+            return;
+        if (socket.close(error))
+            return;
+    }
+}
+
+void OrcaMqttConnection::enqueue_packet(const std::shared_ptr<Connection>& conn,
+                                        std::vector<uint8_t> packet) {
+    if (!conn || packet.empty())
+        return;
+    conn->outbound_packets.emplace_back(std::make_shared<std::vector<uint8_t>>(std::move(packet)));
+    start_async_write(conn);
+}
+
+void OrcaMqttConnection::start_async_write(const std::shared_ptr<Connection>& conn) {
+    if (!conn || conn->write_in_progress || conn->outbound_packets.empty() || stopping.load())
+        return;
+
+    conn->write_in_progress = true;
+    const auto packet = conn->outbound_packets.front();
+    auto on_write = [this, conn](const boost::system::error_code& error, std::size_t) {
+        conn->write_in_progress = false;
+        if (error) {
+            if (!stopping.load())
+                conn->terminal_error = error;
+            conn->io_context.stop();
+            return;
+        }
+        conn->outbound_packets.pop_front();
+        start_async_write(conn);
+    };
+    if (conn->wss) {
+        conn->wss->binary(true);
+        conn->wss->async_write(boost::asio::buffer(*packet), std::move(on_write));
+    } else if (conn->ws) {
+        conn->ws->binary(true);
+        conn->ws->async_write(boost::asio::buffer(*packet), std::move(on_write));
+    } else {
+        conn->write_in_progress = false;
+        conn->outbound_packets.pop_front();
+    }
+}
+
+void OrcaMqttConnection::start_async_read(const std::shared_ptr<Connection>& conn) {
+    if (!conn || stopping.load())
+        return;
+
+    auto on_read = [this, conn](const boost::system::error_code& error, std::size_t) {
+        if (error) {
+            if (!stopping.load())
+                conn->terminal_error = error;
+            conn->io_context.stop();
+            return;
+        }
+        const std::string packet = boost::beast::buffers_to_string(conn->read_buffer.data());
+        conn->read_buffer.consume(conn->read_buffer.size());
+        handle_packet(packet);
+        start_async_read(conn);
+    };
+    if (conn->wss)
+        conn->wss->async_read(conn->read_buffer, std::move(on_read));
+    else if (conn->ws)
+        conn->ws->async_read(conn->read_buffer, std::move(on_read));
+}
+
+void OrcaMqttConnection::schedule_keepalive(const std::shared_ptr<Connection>& conn) {
+    const int keepalive = current_config.keepalive_seconds;
+    if (!conn || keepalive <= 0 || stopping.load())
+        return;
+
+    conn->keepalive_timer.expires_after(std::chrono::seconds(std::max(1, keepalive / 2)));
+    conn->keepalive_timer.async_wait([this, conn](const boost::system::error_code& error) {
+        if (error || stopping.load())
+            return;
+        enqueue_packet(conn, make_ping_packet());
+        schedule_keepalive(conn);
+    });
+}
+
+void OrcaMqttConnection::post_packet(const std::shared_ptr<Connection>& conn,
+                                     std::vector<uint8_t> packet) {
+    if (!conn || packet.empty())
+        return;
+    boost::asio::post(conn->io_context, [this, conn, packet = std::move(packet)]() mutable {
+        if (!stopping.load())
+            enqueue_packet(conn, std::move(packet));
+    });
+}
+
 std::size_t OrcaMqttConnection::ws_read(Connection& conn, boost::beast::flat_buffer& buffer,
                                         boost::system::error_code& ec) {
     if (conn.wss)
@@ -321,14 +437,6 @@ std::size_t OrcaMqttConnection::ws_read(Connection& conn, boost::beast::flat_buf
     return 0;
 }
 
-void OrcaMqttConnection::ws_close(Connection& conn) {
-    boost::system::error_code close_error;
-    if (conn.wss)
-        conn.wss->close(boost::beast::websocket::close_code::normal, close_error);
-    else if (conn.ws)
-        conn.ws->close(boost::beast::websocket::close_code::normal, close_error);
-}
-
 void OrcaMqttConnection::ws_handshake(Connection& conn, const Config& config, const Endpoint& endpoint) {
     const auto results = conn.resolver.resolve(endpoint.host, endpoint.port);
 
@@ -415,12 +523,7 @@ bool OrcaMqttConnection::send_request(const std::string& dev_id, const std::stri
             return true;
         }
     }
-    try {
+    post_packet(conn, make_publish_packet(request_topic(dev_id), payload));
-        // ws_write() serialises the write via write_mutex; do not lock it here.
-        ws_write(*conn, make_publish_packet(request_topic(dev_id), payload));
-    } catch (const std::exception&) {
-        return false;
-    }
     return true;
 }
 
@@ -429,10 +532,20 @@ void OrcaMqttConnection::connect_and_read() {
     {
         std::lock_guard<std::mutex> lock(connection_mutex);
         active_connection = connection;
-        if (stopping.load())
+        if (stopping.load()) {
+            active_connection.reset();
             return;
         }
+    }
 
+    auto clear_connection = [this, connection] {
+        close_connection(*connection);
+        std::lock_guard<std::mutex> lock(connection_mutex);
+        if (active_connection == connection)
+            active_connection.reset();
+    };
+
+    try {
         Endpoint endpoint;
         if (!parse_endpoint(current_config.url, endpoint)) {
             throw std::runtime_error("invalid Orca Cloud WebSocket endpoint");
@@ -485,38 +598,27 @@ void OrcaMqttConnection::connect_and_read() {
             acknowledged_subscriptions.clear();
             pending_subscribe_packets.clear();
         }
+        connection->async_session_started.store(true);
         notify_state(true);
         reconnect_delay_seconds.store(1); // a fresh CONNACK resets the backoff
-    send_current_subscriptions(*connection);
+        send_current_subscriptions(connection);
-    std::chrono::steady_clock::time_point next_ping = std::chrono::steady_clock::now() + std::chrono::seconds(30);
+        start_async_read(connection);
-
+        schedule_keepalive(connection);
-    while (!stopping.load()) {
+    const std::size_t handlers_run = connection->io_context.run();
-        send_pending_subscriptions(*connection);
+    if (handlers_run == 0 && !stopping.load())
-        // Keepalive is driven every iteration, not only from the read-timeout branch:
+        throw std::runtime_error("Orca MQTT event loop stopped unexpectedly");
-        // a printer pushing faster than the 1s read deadline would otherwise keep the
+        if (connection->terminal_error && !stopping.load())
-        // read hot and the broker would drop us at 1.5 x keepalive.
+            throw boost::system::system_error(connection->terminal_error, "read Orca MQTT message");
-        if (std::chrono::steady_clock::now() >= next_ping) {
+        clear_connection();
-            ws_write(*connection, make_ping_packet());
-            next_ping = std::chrono::steady_clock::now() + std::chrono::seconds(30);
-        }
-        buffer.consume(buffer.size());
-        expires_after(*connection, std::chrono::seconds(1));
-        boost::system::error_code error;
-        ws_read(*connection, buffer, error);
-        if (error == boost::beast::error::timeout)
-            continue;
-        if (error) {
-            throw boost::system::system_error(error, "read Orca MQTT message");
-        }
-        handle_packet(boost::beast::buffers_to_string(buffer.data()));
-    }
-
-    ws_close(*connection);
         if (!stopping.load())
             notify_state(false);
+    } catch (...) {
+        clear_connection();
+        throw;
+    }
 }
 
-void OrcaMqttConnection::send_current_subscriptions(Connection& conn) {
+void OrcaMqttConnection::send_current_subscriptions(const std::shared_ptr<Connection>& conn) {
     std::vector<std::string> topics;
     {
         std::lock_guard<std::mutex> lock(mutex);
@@ -532,11 +634,11 @@ void OrcaMqttConnection::send_current_subscriptions(Connection& conn) {
             std::lock_guard<std::mutex> lock(mutex);
             pending_subscribe_packets[packet_id] = topic;
         }
-        ws_write(conn, make_subscribe_packet(packet_id, topic, 1));
+        enqueue_packet(conn, make_subscribe_packet(packet_id, topic, 1));
     }
 }
 
-void OrcaMqttConnection::send_pending_subscriptions(Connection& conn) {
+void OrcaMqttConnection::send_pending_subscriptions(const std::shared_ptr<Connection>& conn) {
     std::vector<std::string> subscribe_topics;
     std::vector<std::string> unsubscribe_topics;
     {
@@ -552,11 +654,11 @@ void OrcaMqttConnection::send_pending_subscriptions(Connection& conn) {
             std::lock_guard<std::mutex> lock(mutex);
             pending_subscribe_packets[packet_id] = topic;
         }
-        ws_write(conn, make_subscribe_packet(packet_id, topic, 1));
+        enqueue_packet(conn, make_subscribe_packet(packet_id, topic, 1));
     }
     for (const std::string& topic : unsubscribe_topics) {
         const uint16_t packet_id = next_packet_id++;
-        ws_write(conn, make_unsubscribe_packet(packet_id, topic));
+        enqueue_packet(conn, make_unsubscribe_packet(packet_id, topic));
     }
 }
 

src/slic3r/Utils/OrcaMqttConnection.hpp

@@ -96,19 +96,23 @@ private:
     static std::vector<uint8_t> make_ping_packet();
 
     // Transport dispatch: each forwards to conn.wss (TLS) or conn.ws (plaintext).
-    void        ws_write(Connection& conn, const std::vector<uint8_t>& packet); // locks write_mutex
+    // These synchronous operations are called only by the MQTT worker during
+    // connection setup. Once the MQTT session is established, all socket I/O is
+    // asynchronous and owned by that worker's io_context.
+    void        ws_write(Connection& conn, const std::vector<uint8_t>& packet);
     std::size_t ws_read(Connection& conn, boost::beast::flat_buffer& buffer, boost::system::error_code& ec);
     void        ws_handshake(Connection& conn, const Config& config, const Endpoint& endpoint);
-    void        ws_close(Connection& conn);
+    void        close_connection(Connection& conn);
-    // Emit a queued SUBSCRIBE/UNSUBSCRIBE on the live socket right now (from the
+    void        enqueue_packet(const std::shared_ptr<Connection>& conn, std::vector<uint8_t> packet);
-    // caller thread), so a selection change is applied without waiting for the
+    void        start_async_write(const std::shared_ptr<Connection>& conn);
-    // blocking read loop to next return. No-op if no CONNACKed socket exists yet
+    void        start_async_read(const std::shared_ptr<Connection>& conn);
-    // (the worker sends the set on connect). The WebSocket is never dropped for a
+    void        schedule_keepalive(const std::shared_ptr<Connection>& conn);
-    // subscription change.
+    void        post_packet(const std::shared_ptr<Connection>& conn, std::vector<uint8_t> packet);
+    // Ask the MQTT worker to emit subscription changes on its own io_context.
     void flush_subscription_change();
     void connect_and_read();
-    void send_current_subscriptions(Connection& conn);
+    void send_current_subscriptions(const std::shared_ptr<Connection>& conn);
-    void send_pending_subscriptions(Connection& conn);
+    void send_pending_subscriptions(const std::shared_ptr<Connection>& conn);
     void handle_packet(const std::string& packet);
     void notify_state(bool is_now_connected);
     void run();
@@ -122,7 +126,6 @@ private:
     std::thread worker;
     std::mutex mutex;
     std::mutex connection_mutex;
-    std::mutex write_mutex; // serialises every websocket write (worker + caller threads)
     std::shared_ptr<Connection> active_connection;
     std::condition_variable initial_cv;
     std::condition_variable state_cv;

tests/slic3rutils/orca_mqtt_mock_broker.hpp

@@ -127,6 +127,9 @@ public:
     // MQTT CONNECTs seen; increments again after a reconnect.
     int connect_count() const { return m_connect_count.load(); }
 
+    // MQTT PINGREQs seen while the client has no other traffic.
+    int ping_count() const { return m_ping_count.load(); }
+
 private:
     void run()
     {
@@ -231,6 +234,7 @@ private:
             return true;
         }
         case 0xc0: // PINGREQ
+            ++m_ping_count;
             write_packet(stream, {0xd0, 0x00});
             return true;
         case 0xe0: // DISCONNECT
@@ -277,8 +281,8 @@ private:
     }
 
     // Every write - the worker's own replies and push_report() from the test
-    // thread - is serialised by m_mutex. beast permits a writer while the worker
+    // thread - is serialised by m_mutex. The production client keeps all of its
-    // is blocked in read(), which is the same arrangement OrcaMqttConnection uses.
+    // WebSocket operations on its MQTT worker instead.
     void write_packet(ws::stream<beast::tcp_stream>& stream, const std::vector<std::uint8_t>& packet)
     {
         std::lock_guard<std::mutex> lock(m_mutex);
@@ -294,11 +298,14 @@ private:
         m_stream_ready = false;
         boost::system::error_code ec;
         auto&                     socket = beast::get_lowest_layer(*m_stream).socket();
-        socket.cancel(ec);
+        if (socket.cancel(ec))
+            return;
         // shutdown() before close() is what actually wakes a blocking read on the
         // worker thread; close() alone does not on POSIX.
-        socket.shutdown(tcp::socket::shutdown_both, ec);
+        if (socket.shutdown(tcp::socket::shutdown_both, ec))
-        socket.close(ec);
+            return;
+        if (socket.close(ec))
+            return;
     }
 
     const bool                                   m_refuse_auth;
@@ -308,6 +315,7 @@ private:
     std::thread                                  m_thread;
     std::atomic_bool                             m_stopping{false};
     std::atomic<int>                             m_connect_count{0};
+    std::atomic<int>                             m_ping_count{0};
     mutable std::mutex                           m_mutex;
     std::optional<ws::stream<beast::tcp_stream>> m_stream;      // guarded by m_mutex
     bool                                         m_stream_ready = false; // guarded by m_mutex

tests/slic3rutils/test_orca_mqtt_connection.cpp

@@ -182,6 +182,20 @@ static void run_round_trip(bool use_tls_flag_only) {
 TEST_CASE("OrcaMqtt round-trip — LAN-style config",   "[OrcaMqtt][.integration]") { run_round_trip(false); }
 TEST_CASE("OrcaMqtt round-trip — cloud-style config", "[OrcaMqtt][.integration]") { run_round_trip(true);  }
 
+TEST_CASE("OrcaMqtt keepalive runs while the connection is idle", "[OrcaMqtt][.integration]") {
+    orca_mqtt_test::MockBroker broker;
+    OrcaMqttConnection conn;
+    OrcaMqttConnection::Config cfg;
+    cfg.url = broker.ws_url();
+    cfg.keepalive_seconds = 2;
+
+    REQUIRE(conn.start(cfg, [](const std::string&, const std::string&) {}, [](bool, bool) {}));
+    for (int i = 0; i < 200 && broker.ping_count() == 0; ++i)
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    CHECK(broker.ping_count() > 0);
+    conn.stop();
+}
+
 TEST_CASE("OrcaMqtt reconnects and re-subscribes after a socket drop", "[OrcaMqtt][.integration]") {
     orca_mqtt_test::MockBroker broker;
     OrcaMqttConnection conn;