Fix Moonraker status stream hanging app exit and missing dead printers (#16604)

The status stream read the websocket synchronously. Beast's
expires_after() only applies to async operations, so when a printer
stopped sending, ws.read() blocked forever, and stop_status_stream()
joined that thread when the app closed. On Windows the stuck process
also froze the next launch through the shared WebView2 browser.

Run the whole stream on one io_context owned by the stream thread, and
have stop_status_stream() post an abort to it. A stop on an open
websocket sends a close frame and waits up to 1 s for the answer.
Keep-alive pings detect a dead printer, and the stream reconnects until
stopped, with the backoff capped at 32 s. When a quiet printer still
answers pings, its last status is sent again. The stream starts under
connect_mutex after a staleness check, and a stop from the stream's own
callback no longer joins its own thread.

The Moonraker tests move into test_moonraker_printer_agent.cpp.

src/slic3r/Utils/MoonrakerPrinterAgent.cpp

@@ -5,6 +5,7 @@
 #include "bambu_networking.hpp"
 #include "libslic3r/Preset.hpp"
 #include "libslic3r/PresetBundle.hpp"
+#include "libslic3r/Utils.hpp"
 #include "slic3r/GUI/GUI_App.hpp"
 #include "slic3r/GUI/I18N.hpp"
 #include "slic3r/GUI/DeviceManager.hpp"
@@ -19,14 +20,19 @@
 #include <boost/algorithm/string/trim.hpp>
 #include <boost/asio/buffer.hpp>
 #include <boost/asio/connect.hpp>
+#include <boost/asio/error.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/asio/steady_timer.hpp>
 #include <boost/beast/core.hpp>
 #include <boost/beast/core/tcp_stream.hpp>
 #include <boost/beast/http/field.hpp>
 #include <boost/beast/core/flat_buffer.hpp>
 #include <boost/beast/core/error.hpp>
 #include <boost/beast/core/buffers_to_string.hpp>
+#include <boost/beast/core/stream_traits.hpp>
+#include <boost/beast/core/string_type.hpp>
 #include <boost/beast/websocket.hpp>
 #include <boost/beast/websocket/stream.hpp>
 #include <boost/beast/websocket/stream_base.hpp>
@@ -37,6 +43,7 @@
 #include <boost/filesystem/operations.hpp>
 #include <boost/log/trivial.hpp>
 #include <algorithm>
+#include <atomic>
 #include <chrono>
 #include <cstdint>
 #include <cctype>
@@ -120,6 +127,31 @@ std::string map_moonraker_state(std::string state)
     return "IDLE";
 }
 
+// Runs one asynchronous operation to completion, because synchronous Beast calls ignore timeouts and cannot be aborted
+template<class Initiate>
+beast::error_code run_until_done(net::io_context& ioc, const std::atomic<bool>& stop, Initiate&& initiate)
+{
+    // An abort that runs before the operation starts finds nothing to cancel
+    if (stop.load()) {
+        return net::error::operation_aborted;
+    }
+    beast::error_code result;
+    bool              done = false;
+    initiate([&](beast::error_code ec, auto&&...) {
+        result = ec;
+        done   = true;
+    });
+    ioc.restart();
+    while (!done && ioc.run_one() != 0) {} // run() never returns while the websocket idle timer is armed
+    return done ? result : beast::error_code{net::error::operation_aborted};
+}
+
+uint64_t steady_now_ms()
+{
+    return static_cast<uint64_t>(
+        std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count());
+}
+
 } // namespace
 
 namespace Slic3r {
@@ -1298,6 +1330,8 @@ bool MoonrakerPrinterAgent::fetch_object_list(const std::string&     base_url,
     }
     http.timeout_connect(5)
         .timeout_max(10)
+        // Gives up once the status stream, the only caller, stops
+        .on_progress([this](Http::Progress, bool& cancel) { cancel = ws_stop.load(); })
         .on_complete([&](std::string body, unsigned status) {
             if (status == 200) {
                 response_body = body;
@@ -1468,19 +1502,47 @@ void MoonrakerPrinterAgent::start_status_stream(const std::string& dev_id, const
         return;
     }
 
+    std::lock_guard<std::mutex> lock(ws_abort_mutex);
     ws_stop.store(false);
-    ws_thread = std::thread([this, dev_id, base_url, api_key]() { run_status_stream(dev_id, base_url, api_key); });
+    ws_thread = std::thread(
+        [this, dev_id, base_url, api_key, timings = ws_timings]() { run_status_stream(dev_id, base_url, api_key, timings); });
 }
 
 void MoonrakerPrinterAgent::stop_status_stream()
 {
-    ws_stop.store(true);
-    if (ws_thread.joinable()) {
-        ws_thread.join();
+    std::thread stopped;
+    {
+        std::lock_guard<std::mutex> lock(ws_abort_mutex);
+        ws_stop.store(true);
+        if (ws_abort) {
+            ws_abort();
+        }
+        // A callback on the stream thread can stop the stream, which then ends on its own and is joined later, outside connect_mutex
+        if (ws_thread.get_id() != std::this_thread::get_id()) {
+            stopped = std::move(ws_thread);
+        }
+    }
+    if (stopped.joinable()) {
+        stopped.join();
     }
 }
 
-void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string base_url, std::string api_key)
+void MoonrakerPrinterAgent::join_stopped_status_stream()
+{
+    std::thread stopped;
+    {
+        std::lock_guard<std::mutex> lock(ws_abort_mutex);
+        // A running stream belongs to a newer connection, so only one that is already stopping is joined
+        if (ws_stop.load() && ws_thread.get_id() != std::this_thread::get_id()) {
+            stopped = std::move(ws_thread);
+        }
+    }
+    if (stopped.joinable()) {
+        stopped.join();
+    }
+}
+
+void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string base_url, std::string api_key, StatusStreamTimings timings)
 {
     WsEndpoint endpoint;
     if (!parse_ws_endpoint(base_url, endpoint)) {
@@ -1492,25 +1554,108 @@ void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string ba
         return;
     }
 
+    net::io_context       ioc;
+    net::steady_timer     backoff{ioc};
+    std::function<void()> abort_connection; // Set while a connection attempt runs
+    const auto            run = [&](auto&& initiate) { return run_until_done(ioc, ws_stop, initiate); };
+
+    // Registered under the lock stop_status_stream() takes, so a stop either finds this abort or has already set ws_stop
+    {
+        std::lock_guard<std::mutex> lock(ws_abort_mutex);
+        if (ws_stop.load()) {
+            return;
+        }
+        ws_abort = [&] {
+            net::post(ioc, [&] {
+                backoff.cancel();
+                if (abort_connection) {
+                    abort_connection();
+                }
+            });
+        };
+    }
+    const ScopeGuard clear_abort([this] {
+        std::lock_guard<std::mutex> lock(ws_abort_mutex);
+        ws_abort = nullptr;
+    });
+
     // Reconnection logic
     ws_reconnect_requested.store(false); // Reset reconnect flag
-    int       retry_count   = 0;
-    const int max_retries   = 10;
-    const int base_delay_ms = 1000;
+    int retry_count = 0;
 
-    while (!ws_stop.load() && retry_count < max_retries) {
+    while (!ws_stop.load()) {
         bool connection_lost = false; // Flag to distinguish clean shutdown from unexpected disconnect
 
+        tcp::resolver                        resolver{ioc};
+        websocket::stream<beast::tcp_stream> ws{ioc};
+        beast::tcp_stream&                   stream = beast::get_lowest_layer(ws);
+        net::steady_timer                    close_deadline{ioc};
+        bool                                 closing          = false;
+        bool                                 deadline_pending = false;
+        auto                                 last_message     = std::chrono::steady_clock::now();
+        // Sends a close frame if the websocket is open, and drops the socket when the printer has not answered within timings.close
+        const auto close_politely = [&](websocket::close_code code) {
+            if (!ws.is_open() || closing) {
+                return;
+            }
+            closing          = true;
+            deadline_pending = true;
+            close_deadline.expires_after(timings.close);
+            close_deadline.async_wait([&](beast::error_code ec) {
+                deadline_pending = false;
+                if (!ec) {
+                    beast::error_code ignored;
+                    stream.socket().close(ignored);
+                }
+            });
+            ws.async_close(code, [&](beast::error_code) {
+                closing = false;
+                close_deadline.cancel();
+            });
+        };
+        abort_connection = [&] {
+            // Beast reports a closing websocket as not open, and dropping its socket would cut off the close frame
+            if (closing) {
+                return;
+            }
+            if (ws.is_open()) {
+                close_politely(websocket::close_code::going_away);
+            } else {
+                beast::error_code ignored;
+                stream.socket().close(ignored);
+            }
+        };
+        const ScopeGuard clear_connection_abort([&] { abort_connection = nullptr; });
+
         try {
-            net::io_context   ioc;
-            tcp::resolver     resolver{ioc};
-            beast::tcp_stream stream{ioc};
+            beast::error_code connect_ec = net::error::host_not_found;
+            for (const auto& entry : resolver.resolve(endpoint.host, endpoint.port)) {
+                beast::error_code ignored;
+                stream.socket().close(ignored); // A failed attempt leaves the socket open
+                stream.expires_after(std::chrono::seconds(5));
+                connect_ec = run([&](auto&& handler) { stream.async_connect(entry.endpoint(), handler); });
+                if (!connect_ec || ws_stop.load()) {
+                    break;
+                }
+            }
+            if (connect_ec) {
+                throw beast::system_error{connect_ec};
+            }
+            // Clears the connect deadline, which would otherwise close the socket 5 s after connecting
+            stream.expires_never();
 
-            stream.expires_after(std::chrono::seconds(10));
-            auto const results = resolver.resolve(endpoint.host, endpoint.port);
-            stream.connect(results);
-
-            websocket::stream<beast::tcp_stream> ws{std::move(stream)};
+            websocket::stream_base::timeout timeouts{};
+            timeouts.handshake_timeout = timings.handshake;
+            timeouts.idle_timeout      = timings.idle;
+            timeouts.keep_alive_pings  = true;
+            ws.set_option(timeouts);
+            // Beast pings on a fixed schedule, so the heartbeat also checks how long the printer has sent no message
+            ws.control_callback([&](websocket::frame_type kind, beast::string_view) {
+                if (kind == websocket::frame_type::pong && !closing &&
+                    std::chrono::steady_clock::now() - last_message >= timings.heartbeat) {
+                    send_heartbeat_if_due(dev_id, timings.heartbeat);
+                }
+            });
             ws.set_option(websocket::stream_base::decorator([&](websocket::request_type& req) {
                 req.set(http::field::user_agent, "OrcaSlicer");
                 if (!api_key.empty()) {
@@ -1522,7 +1667,9 @@ void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string ba
             if (!endpoint.port.empty() && endpoint.port != "80") {
                 host_header += ":" + endpoint.port;
             }
-            ws.handshake(host_header, endpoint.target);
+            if (const auto ec = run([&](auto&& handler) { ws.async_handshake(host_header, endpoint.target, handler); })) {
+                throw beast::system_error{ec};
+            }
             ws.text(true);
 
             // Send client identification
@@ -1534,7 +1681,10 @@ void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string ba
             identify["params"]["type"]        = "agent";
             identify["params"]["url"]         = "https://github.com/SoftFever/OrcaSlicer";
             identify["id"]                    = 0;
-            ws.write(net::buffer(identify.dump()));
+            const std::string identify_payload = identify.dump();
+            if (const auto ec = run([&](auto&& handler) { ws.async_write(net::buffer(identify_payload), handler); })) {
+                throw beast::system_error{ec};
+            }
 
             std::set<std::string> subscribe_objects = {"print_stats", "virtual_sdcard"};
             std::set<std::string> available_objects;
@@ -1586,28 +1736,17 @@ void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string ba
             }
             subscribe["params"]["objects"] = std::move(objects);
             subscribe["id"]                = 1;
-            ws.write(net::buffer(subscribe.dump()));
+            const std::string subscribe_payload = subscribe.dump();
+            if (const auto ec = run([&](auto&& handler) { ws.async_write(net::buffer(subscribe_payload), handler); })) {
+                throw beast::system_error{ec};
+            }
 
             // Read loop
             while (!ws_stop.load()) {
-                ws.next_layer().expires_after(std::chrono::seconds(2));
                 beast::flat_buffer buffer;
-                beast::error_code  ec;
-                ws.read(buffer, ec);
-                if (ec == beast::error::timeout) {
-                    const auto now_ms = static_cast<uint64_t>(
-                        std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count());
-                    const auto last_ms = ws_last_emit_ms.load();
-                    if (last_ms == 0 || now_ms - last_ms >= 10000) {
-                        nlohmann::json message;
-                        {
-                            std::lock_guard<std::recursive_mutex> lock(payload_mutex);
-                            message = build_print_payload_locked();
-                        }
-                        dispatch_message(dev_id, message.dump());
-                        ws_last_emit_ms.store(now_ms);
-                    }
-                    continue;
+                const auto         ec = run([&](auto&& handler) { ws.async_read(buffer, handler); });
+                if (ws_stop.load()) {
+                    break;
                 }
                 if (ec == websocket::error::closed) {
                     connection_lost = true;
@@ -1618,6 +1757,9 @@ void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string ba
                     connection_lost = true;
                     break;
                 }
+                last_message = std::chrono::steady_clock::now();
+                // A message shows the printer is answering, so the next drop starts the backoff from reconnect_delay
+                retry_count = 0;
                 handle_ws_message(dev_id, beast::buffers_to_string(buffer.data()));
                 // Check if handle_ws_message triggered reconnection request
                 if (ws_reconnect_requested.exchange(false)) {
@@ -1625,20 +1767,18 @@ void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string ba
                     break;
                 }
             }
-
-            beast::error_code ec;
-            ws.close(websocket::close_code::normal, ec);
-
-            // Only reset retry count on clean shutdown (not connection_lost)
-            if (!connection_lost && !ws_stop.load()) {
-                retry_count = 0;
-            }
-
         } catch (const std::exception& e) {
-            BOOST_LOG_TRIVIAL(warning) << "MoonrakerPrinterAgent: websocket disconnected: " << e.what();
+            if (!ws_stop.load()) {
+                BOOST_LOG_TRIVIAL(warning) << "MoonrakerPrinterAgent: websocket disconnected: " << e.what();
+            }
             connection_lost = true;
         }
 
+        close_politely(ws_stop.load() ? websocket::close_code::going_away : websocket::close_code::normal);
+        ioc.restart();
+        // Both handlers refer to this connection, so neither may be left queued for the next one
+        while ((closing || deadline_pending) && ioc.run_one() != 0) {}
+
         // Exit loop on clean shutdown
         if (!connection_lost) {
             break;
@@ -1649,17 +1789,13 @@ void MoonrakerPrinterAgent::run_status_stream(std::string dev_id, std::string ba
             break;
         }
 
-        // Exponential backoff before reconnection
-        int delay_ms = base_delay_ms * (1 << std::min(retry_count, 5));
-        BOOST_LOG_TRIVIAL(info) << "MoonrakerPrinterAgent: Reconnecting in " << delay_ms << "ms (attempt " << (retry_count + 1) << ")";
-        std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
+        // Exponential backoff before reconnection, capped at 32 times reconnect_delay so a printer that comes back is found again
+        const auto delay = timings.reconnect_delay * (1 << std::min(retry_count, 5));
+        BOOST_LOG_TRIVIAL(info) << "MoonrakerPrinterAgent: Reconnecting in " << delay.count() << "ms (attempt " << (retry_count + 1) << ")";
+        backoff.expires_after(delay);
+        run([&](auto&& handler) { backoff.async_wait(handler); });
         retry_count++;
     }
-
-    if (retry_count >= max_retries) {
-        BOOST_LOG_TRIVIAL(error) << "MoonrakerPrinterAgent: Max reconnection attempts reached";
-        dispatch_local_connect(ConnectStatusLost, dev_id, "max_retries");
-    }
 }
 
 void MoonrakerPrinterAgent::handle_ws_message(const std::string& dev_id, const std::string& payload)
@@ -1733,8 +1869,7 @@ void MoonrakerPrinterAgent::handle_ws_message(const std::string& dev_id, const s
     }
 
     if (updated) {
-        const auto now_ms = static_cast<uint64_t>(
-            std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count());
+        const auto now_ms           = steady_now_ms();
         const auto last_dispatch_ms = ws_last_dispatch_ms.load();
 
         // Dispatch if: critical change OR throttle interval elapsed
@@ -1755,6 +1890,24 @@ void MoonrakerPrinterAgent::handle_ws_message(const std::string& dev_id, const s
     }
 }
 
+void MoonrakerPrinterAgent::send_heartbeat_if_due(const std::string& dev_id, std::chrono::milliseconds interval)
+{
+    const auto now_ms  = steady_now_ms();
+    const auto last_ms = ws_last_emit_ms.load();
+    // Re-sends only a status the printer has reported at least once
+    if (last_ms == 0 || now_ms - last_ms < static_cast<uint64_t>(interval.count())) {
+        return;
+    }
+
+    nlohmann::json message;
+    {
+        std::lock_guard<std::recursive_mutex> lock(payload_mutex);
+        message = build_print_payload_locked();
+    }
+    dispatch_message(dev_id, message.dump());
+    ws_last_emit_ms.store(now_ms);
+}
+
 void MoonrakerPrinterAgent::update_status_cache(const nlohmann::json& updates)
 {
     if (!updates.is_object()) {
@@ -2139,8 +2292,15 @@ void MoonrakerPrinterAgent::perform_connection_async(const std::string& dev_id,
             BOOST_LOG_TRIVIAL(warning) << "MoonrakerPrinterAgent: Initial status query failed: " << error_msg;
         }
 
-        // Start WebSocket status stream
-        start_status_stream(dev_id, base_url, api_key);
+        join_stopped_status_stream();
+        // Holds connect_mutex across the staleness check and the start, so a disconnect during the queries above prevents the start
+        {
+            std::lock_guard<std::recursive_mutex> lock(connect_mutex);
+            if (is_stale()) {
+                return;
+            }
+            start_status_stream(dev_id, base_url, api_key);
+        }
 #endif
 
         // Success!

src/slic3r/Utils/MoonrakerPrinterAgent.hpp

@@ -5,6 +5,7 @@
 #include "bambu_networking.hpp"
 
 #include <functional>
+#include <chrono>
 #include <cstdint>
 #include <atomic>
 #include <memory>
@@ -85,6 +86,16 @@ protected:
         bool        use_ssl = false;
     } device_info;
 
+    // Copied when the stream starts, so changes apply to the next start_status_stream()
+    struct StatusStreamTimings
+    {
+        std::chrono::milliseconds handshake{std::chrono::seconds(10)};      // Opening and closing handshakes
+        std::chrono::milliseconds close{std::chrono::seconds(1)};           // Wait for the close frame's answer before dropping the socket
+        std::chrono::milliseconds idle{std::chrono::seconds(30)};           // Silence before giving up, with a ping every half of it
+        std::chrono::milliseconds heartbeat{std::chrono::seconds(10)};      // Silence before the cached status is sent again
+        std::chrono::milliseconds reconnect_delay{std::chrono::seconds(1)}; // First wait, doubling per consecutive failure up to 32x
+    } ws_timings;
+
     // Tray data for AMS payload building
     struct AmsTrayData {
         int         slot_index = 0;      // 0-based slot index
@@ -102,6 +113,8 @@ protected:
     // Methods that derived classes may need to override or access
     virtual bool init_device_info(const std::string& dev_id, const std::string& dev_ip, const std::string& username, const std::string& password, bool use_ssl, const std::string& port);
     virtual bool fetch_device_info(const std::string& base_url, const std::string& api_key, MoonrakerDeviceInfo& info, std::string& error) const;
+    void start_status_stream(const std::string& dev_id, const std::string& base_url, const std::string& api_key);
+    void stop_status_stream();
 
     // State access for derived classes
     mutable std::recursive_mutex       state_mutex;
@@ -133,10 +146,11 @@ private:
     void dispatch_local_connect(int state, const std::string& dev_id, const std::string& msg);
     void dispatch_printer_connected(const std::string& dev_id);
     void dispatch_message(const std::string& dev_id, const std::string& payload);
-    void start_status_stream(const std::string& dev_id, const std::string& base_url, const std::string& api_key);
-    void stop_status_stream();
-    void run_status_stream(std::string dev_id, std::string base_url, std::string api_key);
+    // Joins a stream stopped from its own callback, so a new stream can start without waiting on it under connect_mutex
+    void join_stopped_status_stream();
+    void run_status_stream(std::string dev_id, std::string base_url, std::string api_key, StatusStreamTimings timings);
     void handle_ws_message(const std::string& dev_id, const std::string& payload);
+    void send_heartbeat_if_due(const std::string& dev_id, std::chrono::milliseconds interval);
     void update_status_cache(const nlohmann::json& updates);
     nlohmann::json build_print_payload_locked() const;
 
@@ -196,6 +210,8 @@ private:
     std::atomic<bool>   ws_reconnect_requested{false};  // Flag to trigger reconnection
     std::atomic<uint64_t> ws_last_emit_ms{0};
     std::thread         ws_thread;
+    std::mutex          ws_abort_mutex; // Guards ws_thread and ws_abort
+    std::function<void()> ws_abort; // Interrupts whatever the stream thread is waiting on, from any thread
 
     // Throttling configuration for WebSocket updates
     // Critical changes (state transitions) dispatch immediately; telemetry is throttled

tests/slic3rutils/CMakeLists.txt

@@ -6,7 +6,7 @@ add_executable(${_TEST_NAME}_tests
     test_creality_cfs_match.cpp
     test_dev_mapping.cpp
     test_device_access.cpp
-    test_moonraker_discovery.cpp
+    test_moonraker_printer_agent.cpp
     test_network_discovery.cpp
     test_filament_bitmap_utils.cpp
     test_colour_picker_field.cpp

tests/slic3rutils/test_moonraker_discovery.cpp

@@ -1,51 +0,0 @@
-#include <catch2/catch_test_macros.hpp>
-
-#include <string>
-#include "slic3r/Utils/MoonrakerPrinterAgent.hpp"
-
-using namespace Slic3r;
-
-namespace {
-
-class DiscoveryAgent : public MoonrakerPrinterAgent
-{
-public:
-    using MoonrakerPrinterAgent::MoonrakerPrinterAgent;
-    using MoonrakerPrinterAgent::device_info;
-    mutable unsigned fetch_count = 0;
-
-protected:
-    bool fetch_device_info(const std::string&, const std::string&, MoonrakerDeviceInfo& info, std::string&) const override
-    {
-        ++fetch_count;
-        info.dev_name = "Test printer";
-        return true;
-    }
-};
-
-} // namespace
-
-TEST_CASE("Moonraker discovery waits for a complete device identity", "[MoonrakerDiscovery]")
-{
-    DiscoveryAgent agent("");
-    unsigned announcements = 0;
-    nlohmann::json announcement;
-    agent.set_on_ssdp_msg_fn([&](std::string payload) {
-        ++announcements;
-        announcement = nlohmann::json::parse(payload);
-    });
-
-    agent.start_discovery(true, false);
-    REQUIRE(announcements == 0);
-    REQUIRE(agent.fetch_count == 0);
-
-    agent.device_info.dev_id = "127.0.0.1:7125";
-    agent.device_info.dev_ip = agent.device_info.dev_id;
-    agent.device_info.base_url = "http://127.0.0.1:7125";
-    agent.start_discovery(true, false);
-    REQUIRE(announcements == 1);
-    REQUIRE(agent.fetch_count == 1);
-    REQUIRE(announcement.at("dev_id") == agent.device_info.dev_id);
-    REQUIRE(announcement.at("dev_ip") == agent.device_info.dev_ip);
-    REQUIRE(announcement.at("dev_name") == "Test printer");
-}

tests/slic3rutils/test_moonraker_printer_agent.cpp

@@ -0,0 +1,580 @@
+#include <catch2/catch_message.hpp>
+#include <catch2/catch_test_macros.hpp>
+#include <catch2/generators/catch_generators.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstddef>
+#include <functional>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include <boost/asio/buffer.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/address.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/beast/core/error.hpp>
+#include <boost/beast/core/flat_buffer.hpp>
+#include <boost/beast/http/message.hpp>
+#include <boost/beast/http/read.hpp>
+#include <boost/beast/http/status.hpp>
+#include <boost/beast/http/string_body.hpp>
+#include <boost/beast/http/write.hpp>
+#include <boost/beast/websocket/error.hpp>
+#include <boost/beast/websocket/rfc6455.hpp>
+#include <boost/beast/websocket/stream.hpp>
+#include <nlohmann/json.hpp>
+
+#include "slic3r/Utils/IPrinterAgent.hpp"
+#include "slic3r/Utils/MoonrakerPrinterAgent.hpp"
+
+using namespace Slic3r;
+
+namespace {
+
+namespace beast     = boost::beast;
+namespace http      = beast::http;
+namespace websocket = beast::websocket;
+namespace net       = boost::asio;
+using tcp           = net::ip::tcp;
+using namespace std::chrono_literals;
+
+class DiscoveryAgent : public MoonrakerPrinterAgent
+{
+public:
+    using MoonrakerPrinterAgent::MoonrakerPrinterAgent;
+    using MoonrakerPrinterAgent::device_info;
+    mutable unsigned fetch_count = 0;
+
+protected:
+    bool fetch_device_info(const std::string&, const std::string&, MoonrakerDeviceInfo& info, std::string&) const override
+    {
+        ++fetch_count;
+        info.dev_name = "Test printer";
+        return true;
+    }
+};
+
+class StreamAgent : public MoonrakerPrinterAgent
+{
+public:
+    using MoonrakerPrinterAgent::MoonrakerPrinterAgent;
+    using MoonrakerPrinterAgent::start_status_stream;
+    using MoonrakerPrinterAgent::stop_status_stream;
+    using MoonrakerPrinterAgent::ws_timings;
+
+protected:
+    bool init_device_info(
+        const std::string& dev_id, const std::string& dev_ip, const std::string&, const std::string&, bool, const std::string& port) override
+    {
+        device_info          = MoonrakerDeviceInfo{};
+        device_info.dev_id   = dev_id;
+        device_info.base_url = "http://" + dev_ip + ":" + port;
+        return true;
+    }
+
+    bool fetch_device_info(const std::string&, const std::string&, MoonrakerDeviceInfo& info, std::string&) const override
+    {
+        info.dev_name = "Test printer";
+        return true;
+    }
+};
+
+enum class Behavior {
+    SilentBeforeHandshake,     // Accepts the connection and never answers
+    Silent,                    // Completes the upgrade, then never reads or sends
+    AnswersPings,              // Reports a status, then answers pings and sends nothing else
+    AnswersPingsWithoutStatus, // Answers pings but never reports a status
+    Chatters,                  // Reports a status, then answers pings and sends notifications that carry no status
+    HangsUp,                   // Reports a status, then closes the websocket
+    HangsUpWithoutAnswer,      // Closes the websocket without sending anything
+    IgnoresClose,              // Reports Klippy disconnected, then never answers the close
+    StallsObjectList,          // Never answers the stream's object list request
+    StallsStatusQuery,         // Never answers HTTP requests until released, then hangs up on them
+};
+
+class FakePrinter
+{
+public:
+    explicit FakePrinter(Behavior behavior) : m_behavior(behavior)
+    {
+        m_accept_thread = std::thread([this] { accept_loop(); });
+    }
+
+    ~FakePrinter()
+    {
+        release();
+        m_stopping = true;
+        // Wakes the blocking accept.
+        tcp::socket wake{m_accept_ioc};
+        beast::error_code ec;
+        wake.connect(m_acceptor.local_endpoint(), ec);
+        m_accept_thread.join();
+        for (std::thread& connection : m_connections)
+            connection.join();
+    }
+
+    std::string host() const { return "127.0.0.1"; }
+    std::string port() const { return std::to_string(m_acceptor.local_endpoint().port()); }
+    std::string base_url() const { return "http://" + host() + ":" + port(); }
+
+    // A connection is ready once it reaches the state its behavior waits in.
+    bool wait_for_ready_connections(int count, std::chrono::milliseconds timeout = 10s)
+    {
+        std::unique_lock<std::mutex> lock(m_mutex);
+        return m_changed.wait_for(lock, timeout, [&] { return m_ready >= count; });
+    }
+
+    bool wait_for_http_requests(int count)
+    {
+        std::unique_lock<std::mutex> lock(m_mutex);
+        return m_changed.wait_for(lock, 10s, [&] { return m_http_requests >= count; });
+    }
+
+    // Waits until the printer, once released, has hung up on `count` stalled HTTP requests.
+    bool wait_for_http_hang_ups(int count)
+    {
+        std::unique_lock<std::mutex> lock(m_mutex);
+        return m_changed.wait_for(lock, 10s, [&] { return m_http_hung_up >= count; });
+    }
+
+    int close_frames()
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        return m_close_frames;
+    }
+
+    // Hangs up every connection that is waiting.
+    void release()
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        m_released = true;
+        m_changed.notify_all();
+    }
+
+private:
+    void accept_loop()
+    {
+        for (;;) {
+            // Each connection gets its own io_context, so a connection that runs one never runs another's handlers.
+            auto ioc = std::make_unique<net::io_context>();
+            tcp::socket socket{*ioc};
+            beast::error_code ec;
+            m_acceptor.accept(socket, ec);
+            if (ec || m_stopping)
+                return;
+            m_connections.emplace_back(
+                [this, ioc = std::move(ioc), socket = std::move(socket)]() mutable { serve(*ioc, std::move(socket)); });
+        }
+    }
+
+    void serve(net::io_context& ioc, tcp::socket socket)
+    {
+        if (m_behavior == Behavior::SilentBeforeHandshake) {
+            mark_ready();
+            wait_for_release();
+            return;
+        }
+
+        beast::flat_buffer buffer;
+        http::request<http::string_body> request;
+        beast::error_code ec;
+        http::read(socket, buffer, request, ec);
+        if (ec)
+            return;
+        // The agent also queries the printer's status and object list over plain HTTP.
+        if (!websocket::is_upgrade(request)) {
+            serve_http(socket, request);
+            return;
+        }
+
+        websocket::stream<tcp::socket> ws{std::move(socket)};
+        ws.accept(request, ec);
+        // Without its object list the stream never sends the subscribe request.
+        const int requests = m_behavior == Behavior::StallsObjectList ? 1 : 2;
+        for (int i = 0; i < requests && !ec; ++i) {
+            beast::flat_buffer message;
+            ws.read(message, ec);
+        }
+        if (!ec && (m_behavior == Behavior::AnswersPings || m_behavior == Behavior::Chatters || m_behavior == Behavior::HangsUp)) {
+            const std::string status =
+                nlohmann::json{{"jsonrpc", "2.0"}, {"id", 1}, {"result", {{"status", {{"print_stats", {{"state", "standby"}}}}}}}}.dump();
+            ws.write(net::buffer(status), ec);
+        }
+        if (ec)
+            return;
+
+        switch (m_behavior) {
+        case Behavior::AnswersPings:
+        case Behavior::AnswersPingsWithoutStatus:
+        case Behavior::Chatters:
+            mark_ready();
+            answer_pings(ioc, ws, m_behavior == Behavior::Chatters);
+            break;
+        case Behavior::HangsUp:
+        case Behavior::HangsUpWithoutAnswer:
+            ws.close(websocket::close_code::normal, ec);
+            mark_ready();
+            break;
+        case Behavior::IgnoresClose: {
+            const std::string notification = nlohmann::json{{"jsonrpc", "2.0"}, {"method", "notify_klippy_disconnected"}}.dump();
+            ws.write(net::buffer(notification), ec);
+            mark_ready();
+            wait_for_release();
+            break;
+        }
+        case Behavior::StallsObjectList: answer_pings(ioc, ws, false); break;
+        default: mark_ready(); wait_for_release();
+        }
+    }
+
+    void serve_http(tcp::socket& socket, const http::request<http::string_body>& request)
+    {
+        {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            ++m_http_requests;
+            m_changed.notify_all();
+        }
+        if (m_behavior == Behavior::StallsObjectList) {
+            mark_ready();
+            wait_for_release();
+            return;
+        }
+        if (m_behavior == Behavior::StallsStatusQuery) {
+            wait_for_release();
+            beast::error_code ec;
+            socket.close(ec);
+            std::lock_guard<std::mutex> lock(m_mutex);
+            ++m_http_hung_up;
+            m_changed.notify_all();
+            return;
+        }
+        http::response<http::string_body> response{http::status::not_found, request.version()};
+        response.prepare_payload();
+        beast::error_code ec;
+        http::write(socket, response, ec);
+    }
+
+    // Keeps a read pending, which answers pings, until released or closed, and with `chatter` also sends a notification every 50 ms.
+    void answer_pings(net::io_context& ioc, websocket::stream<tcp::socket>& ws, bool chatter)
+    {
+        const std::string notification = nlohmann::json{{"jsonrpc", "2.0"},
+                                                        {"method", "notify_proc_stat_update"},
+                                                        {"params", nlohmann::json::array({nlohmann::json::object()})}}
+                                             .dump();
+        beast::flat_buffer message;
+        bool reading                    = true;
+        bool writing                    = false;
+        std::function<void()> read_next = [&] {
+            ws.async_read(message, [&](beast::error_code ec, std::size_t) {
+                message.clear();
+                if (ec == websocket::error::closed) {
+                    std::lock_guard<std::mutex> lock(m_mutex);
+                    ++m_close_frames;
+                }
+                if (ec)
+                    reading = false;
+                else
+                    read_next();
+            });
+        };
+        read_next();
+
+        auto next_notification = std::chrono::steady_clock::now();
+        while (reading && !wait_for_release(10ms)) {
+            if (chatter && !writing && std::chrono::steady_clock::now() >= next_notification) {
+                writing = true;
+                next_notification += 50ms;
+                ws.async_write(net::buffer(notification), [&](beast::error_code, std::size_t) { writing = false; });
+            }
+            ioc.restart();
+            ioc.run_for(20ms);
+        }
+        beast::error_code ignored;
+        ws.next_layer().close(ignored);
+        while (reading || writing) {
+            ioc.restart();
+            ioc.run_for(20ms);
+        }
+    }
+
+    void mark_ready()
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        ++m_ready;
+        m_changed.notify_all();
+    }
+
+    void wait_for_release()
+    {
+        std::unique_lock<std::mutex> lock(m_mutex);
+        m_changed.wait(lock, [this] { return m_released; });
+    }
+
+    bool wait_for_release(std::chrono::milliseconds timeout)
+    {
+        std::unique_lock<std::mutex> lock(m_mutex);
+        return m_changed.wait_for(lock, timeout, [this] { return m_released; });
+    }
+
+    const Behavior m_behavior;
+    net::io_context m_accept_ioc;
+    tcp::acceptor m_acceptor{m_accept_ioc, {net::ip::make_address("127.0.0.1"), 0}};
+    std::atomic<bool> m_stopping{false};
+    std::thread m_accept_thread;
+    std::vector<std::thread> m_connections;
+    std::mutex m_mutex;
+    std::condition_variable m_changed;
+    int m_ready         = 0;
+    int m_http_requests = 0;
+    int m_http_hung_up  = 0;
+    int m_close_frames  = 0;
+    bool m_released     = false;
+};
+
+// Records when the agent dispatches a status report.
+class Reports
+{
+public:
+    void attach(StreamAgent& agent)
+    {
+        agent.set_on_local_message_fn([this](std::string, std::string) {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            m_times.push_back(std::chrono::steady_clock::now());
+            m_changed.notify_all();
+        });
+    }
+
+    bool wait_for(std::size_t count)
+    {
+        std::unique_lock<std::mutex> lock(m_mutex);
+        return m_changed.wait_for(lock, 10s, [&] { return m_times.size() >= count; });
+    }
+
+    std::vector<std::chrono::steady_clock::time_point> times()
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        return m_times;
+    }
+
+private:
+    std::mutex m_mutex;
+    std::condition_variable m_changed;
+    std::vector<std::chrono::steady_clock::time_point> m_times;
+};
+
+// A printer and a stream agent connected to it.
+struct StreamFixture
+{
+    explicit StreamFixture(Behavior behavior) : printer(behavior) {}
+
+    // Runs before the members are destroyed, so the agent's destructor never joins a stream the printer still holds open.
+    ~StreamFixture() { printer.release(); }
+
+    void start() { agent.start_status_stream("printer", printer.base_url(), ""); }
+
+    // Stops the stream, giving it `limit` to return before the printer hangs up on it.
+    bool stops_within(std::chrono::milliseconds limit)
+    {
+        auto stopped       = std::async(std::launch::async, [this] { agent.stop_status_stream(); });
+        const bool in_time = stopped.wait_for(limit) == std::future_status::ready;
+        printer.release();
+        stopped.wait();
+        return in_time;
+    }
+
+    FakePrinter printer;
+    StreamAgent agent{""};
+};
+
+} // namespace
+
+TEST_CASE("Moonraker discovery waits for a complete device identity", "[MoonrakerPrinterAgent][Discovery]")
+{
+    DiscoveryAgent agent("");
+    unsigned announcements = 0;
+    nlohmann::json announcement;
+    agent.set_on_ssdp_msg_fn([&](std::string payload) {
+        ++announcements;
+        announcement = nlohmann::json::parse(payload);
+    });
+
+    agent.start_discovery(true, false);
+    REQUIRE(announcements == 0);
+    REQUIRE(agent.fetch_count == 0);
+
+    agent.device_info.dev_id = "127.0.0.1:7125";
+    agent.device_info.dev_ip = agent.device_info.dev_id;
+    agent.device_info.base_url = "http://127.0.0.1:7125";
+    agent.start_discovery(true, false);
+    REQUIRE(announcements == 1);
+    REQUIRE(agent.fetch_count == 1);
+    REQUIRE(announcement.at("dev_id") == agent.device_info.dev_id);
+    REQUIRE(announcement.at("dev_ip") == agent.device_info.dev_ip);
+    REQUIRE(announcement.at("dev_name") == "Test printer");
+}
+
+TEST_CASE("Stopping the status stream gives a silent printer no more than its timeouts", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    // An open websocket gets the 200 ms close timeout, and a stalled object list request up to 1 s of curl progress to see the stop.
+    const auto [behavior, limit] = GENERATE(table<Behavior, std::chrono::milliseconds>({
+        {Behavior::SilentBeforeHandshake, 500ms},
+        {Behavior::Silent, 1000ms},
+        {Behavior::StallsObjectList, 2000ms},
+    }));
+    CAPTURE(static_cast<int>(behavior));
+
+    StreamFixture f(behavior);
+    f.agent.ws_timings.close = 200ms;
+    f.start();
+    REQUIRE(f.printer.wait_for_ready_connections(1));
+
+    CHECK(f.stops_within(limit));
+}
+
+TEST_CASE("Stopping the status stream sends the printer a close frame", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    // A stop while the stream reads, and one while it waits on the object list request.
+    const Behavior behavior = GENERATE(Behavior::AnswersPings, Behavior::StallsObjectList);
+    CAPTURE(static_cast<int>(behavior));
+
+    StreamFixture f(behavior);
+    f.start();
+    REQUIRE(f.printer.wait_for_ready_connections(1));
+
+    CHECK(f.stops_within(3s));
+    CHECK(f.printer.close_frames() == 1);
+}
+
+TEST_CASE("Stopping the status stream does not wait out the reconnect delay", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    StreamFixture f(Behavior::HangsUp);
+    f.agent.ws_timings.reconnect_delay = 10s;
+    f.start();
+    REQUIRE(f.printer.wait_for_ready_connections(1));
+    // Lets the stream reach its reconnect wait.
+    std::this_thread::sleep_for(200ms);
+
+    CHECK(f.stops_within(3s));
+}
+
+TEST_CASE("The status stream reconnects when the printer stops answering", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    // An unanswered close ends at whichever of the handshake and close timeouts is shorter.
+    const auto [behavior, handshake, close] = GENERATE(table<Behavior, std::chrono::milliseconds, std::chrono::milliseconds>({
+        {Behavior::Silent, 300ms, 1000ms},
+        {Behavior::IgnoresClose, 300ms, 1000ms},
+        {Behavior::IgnoresClose, 1000ms, 200ms},
+    }));
+    CAPTURE(static_cast<int>(behavior), handshake.count(), close.count());
+
+    StreamFixture f(behavior);
+    f.agent.ws_timings.handshake       = handshake;
+    f.agent.ws_timings.close           = close;
+    f.agent.ws_timings.idle            = 400ms;
+    f.agent.ws_timings.reconnect_delay = 20ms;
+    f.start();
+
+    CHECK(f.printer.wait_for_ready_connections(2));
+    CHECK(f.stops_within(3s));
+}
+
+TEST_CASE("The status stream reconnects at the shortest delay while the printer answers", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    StreamFixture f(Behavior::HangsUp);
+    f.agent.ws_timings.reconnect_delay = 50ms;
+    f.start();
+
+    // Twelve 50 ms waits; a backoff that kept doubling would need over 10 s.
+    CHECK(f.printer.wait_for_ready_connections(12, 3s));
+    CHECK(f.stops_within(3s));
+}
+
+TEST_CASE("The status stream keeps retrying a printer that never answers", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    StreamFixture f(Behavior::HangsUpWithoutAnswer);
+    f.agent.ws_timings.reconnect_delay = 1ms;
+    f.start();
+
+    // The backoff stops doubling at 32 ms, so 20 attempts take well under the wait.
+    CHECK(f.printer.wait_for_ready_connections(20));
+    CHECK(f.stops_within(3s));
+}
+
+TEST_CASE("A callback on the stream thread can stop the status stream", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    std::promise<void> stopped;
+    std::once_flag stopped_once;
+    StreamFixture f(Behavior::AnswersPings);
+    f.agent.set_on_local_message_fn([&](std::string, std::string) {
+        std::call_once(stopped_once, [&] {
+            f.agent.stop_status_stream();
+            stopped.set_value();
+        });
+    });
+    f.start();
+
+    REQUIRE(stopped.get_future().wait_for(5s) == std::future_status::ready);
+    CHECK(f.stops_within(3s));
+}
+
+TEST_CASE("A quiet printer that is still up keeps reporting its status", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    Reports reports;
+    StreamFixture f(Behavior::AnswersPings);
+    f.agent.ws_timings.idle      = 400ms;
+    f.agent.ws_timings.heartbeat = 300ms;
+    reports.attach(f.agent);
+    f.start();
+
+    // The status the printer sent, then two heartbeats.
+    REQUIRE(reports.wait_for(3));
+    const auto times = reports.times();
+    // The steady clock is compared in whole milliseconds, so allow 1 ms of rounding.
+    CHECK(times[1] - times[0] >= 299ms);
+    CHECK(times[2] - times[1] >= 299ms);
+    CHECK(f.stops_within(3s));
+}
+
+TEST_CASE("The status stream sends no heartbeat while the printer has nothing new to report", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    const Behavior behavior = GENERATE(Behavior::Chatters, Behavior::AnswersPingsWithoutStatus);
+    CAPTURE(static_cast<int>(behavior));
+
+    Reports reports;
+    StreamFixture f(behavior);
+    f.agent.ws_timings.idle      = 400ms;
+    f.agent.ws_timings.heartbeat = 800ms;
+    reports.attach(f.agent);
+    f.start();
+    REQUIRE(f.printer.wait_for_ready_connections(1));
+
+    // The chattering printer's status, and nothing from the one that never sent a status.
+    const std::size_t expected = behavior == Behavior::Chatters ? 1 : 0;
+    std::this_thread::sleep_for(1500ms);
+    CHECK(reports.times().size() == expected);
+    CHECK(f.stops_within(3s));
+}
+
+TEST_CASE("A disconnect during the connection queries keeps the status stream from starting", "[MoonrakerPrinterAgent][StatusStream]")
+{
+    StreamFixture f(Behavior::StallsStatusQuery);
+    PrinterConnectionParams params;
+    params.dev_id = "printer";
+    params.host   = f.printer.host();
+    params.port   = f.printer.port();
+    f.agent.connect_printer(params);
+    REQUIRE(f.printer.wait_for_http_requests(1));
+
+    f.agent.disconnect_printer();
+    f.printer.release();
+    // The query has returned once the printer hangs up on it, and a stream would connect right after.
+    REQUIRE(f.printer.wait_for_http_hang_ups(1));
+    CHECK_FALSE(f.printer.wait_for_ready_connections(1, 500ms));
+}