From da4d766d467cb90274393523c8ba68dea74bb3c2 Mon Sep 17 00:00:00 2001 From: Kris Austin Date: Sun, 18 Oct 2026 19:33:58 -0500 Subject: [PATCH] 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 | 280 +++++++-- src/slic3r/Utils/MoonrakerPrinterAgent.hpp | 22 +- tests/slic3rutils/CMakeLists.txt | 2 +- .../slic3rutils/test_moonraker_discovery.cpp | 51 -- .../test_moonraker_printer_agent.cpp | 580 ++++++++++++++++++ 5 files changed, 820 insertions(+), 115 deletions(-) delete mode 100644 tests/slic3rutils/test_moonraker_discovery.cpp create mode 100644 tests/slic3rutils/test_moonraker_printer_agent.cpp diff --git a/src/slic3r/Utils/MoonrakerPrinterAgent.cpp b/src/slic3r/Utils/MoonrakerPrinterAgent.cpp index 81a25165a2..8494eb5cfd 100644 --- a/src/slic3r/Utils/MoonrakerPrinterAgent.cpp +++ b/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 #include #include +#include #include #include +#include +#include #include #include #include #include #include #include +#include +#include #include #include #include @@ -37,6 +43,7 @@ #include #include #include +#include #include #include #include @@ -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 +beast::error_code run_until_done(net::io_context& ioc, const std::atomic& 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( + std::chrono::duration_cast(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 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 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 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 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 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 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 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 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 subscribe_objects = {"print_stats", "virtual_sdcard"}; std::set 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( - std::chrono::duration_cast(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 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( - std::chrono::duration_cast(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(interval.count())) { + return; + } + + nlohmann::json message; + { + std::lock_guard 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 lock(connect_mutex); + if (is_stale()) { + return; + } + start_status_stream(dev_id, base_url, api_key); + } #endif // Success! diff --git a/src/slic3r/Utils/MoonrakerPrinterAgent.hpp b/src/slic3r/Utils/MoonrakerPrinterAgent.hpp index 141574b84d..a7edc476e8 100644 --- a/src/slic3r/Utils/MoonrakerPrinterAgent.hpp +++ b/src/slic3r/Utils/MoonrakerPrinterAgent.hpp @@ -5,6 +5,7 @@ #include "bambu_networking.hpp" #include +#include #include #include #include @@ -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 ws_reconnect_requested{false}; // Flag to trigger reconnection std::atomic ws_last_emit_ms{0}; std::thread ws_thread; + std::mutex ws_abort_mutex; // Guards ws_thread and ws_abort + std::function 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 diff --git a/tests/slic3rutils/CMakeLists.txt b/tests/slic3rutils/CMakeLists.txt index 83632601f7..c9285b955e 100644 --- a/tests/slic3rutils/CMakeLists.txt +++ b/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 diff --git a/tests/slic3rutils/test_moonraker_discovery.cpp b/tests/slic3rutils/test_moonraker_discovery.cpp deleted file mode 100644 index f4df147225..0000000000 --- a/tests/slic3rutils/test_moonraker_discovery.cpp +++ /dev/null @@ -1,51 +0,0 @@ -#include - -#include -#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"); -} diff --git a/tests/slic3rutils/test_moonraker_printer_agent.cpp b/tests/slic3rutils/test_moonraker_printer_agent.cpp new file mode 100644 index 0000000000..32dca2ffd3 --- /dev/null +++ b/tests/slic3rutils/test_moonraker_printer_agent.cpp @@ -0,0 +1,580 @@ +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#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 lock(m_mutex); + return m_changed.wait_for(lock, timeout, [&] { return m_ready >= count; }); + } + + bool wait_for_http_requests(int count) + { + std::unique_lock 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 lock(m_mutex); + return m_changed.wait_for(lock, 10s, [&] { return m_http_hung_up >= count; }); + } + + int close_frames() + { + std::lock_guard lock(m_mutex); + return m_close_frames; + } + + // Hangs up every connection that is waiting. + void release() + { + std::lock_guard 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(); + 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 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 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& request) + { + { + std::lock_guard 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 lock(m_mutex); + ++m_http_hung_up; + m_changed.notify_all(); + return; + } + http::response 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& 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 read_next = [&] { + ws.async_read(message, [&](beast::error_code ec, std::size_t) { + message.clear(); + if (ec == websocket::error::closed) { + std::lock_guard 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 lock(m_mutex); + ++m_ready; + m_changed.notify_all(); + } + + void wait_for_release() + { + std::unique_lock lock(m_mutex); + m_changed.wait(lock, [this] { return m_released; }); + } + + bool wait_for_release(std::chrono::milliseconds timeout) + { + std::unique_lock 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 m_stopping{false}; + std::thread m_accept_thread; + std::vector 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 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 lock(m_mutex); + return m_changed.wait_for(lock, 10s, [&] { return m_times.size() >= count; }); + } + + std::vector times() + { + std::lock_guard lock(m_mutex); + return m_times; + } + +private: + std::mutex m_mutex; + std::condition_variable m_changed; + std::vector 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::SilentBeforeHandshake, 500ms}, + {Behavior::Silent, 1000ms}, + {Behavior::StallsObjectList, 2000ms}, + })); + CAPTURE(static_cast(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(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::Silent, 300ms, 1000ms}, + {Behavior::IgnoresClose, 300ms, 1000ms}, + {Behavior::IgnoresClose, 1000ms, 200ms}, + })); + CAPTURE(static_cast(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 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(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)); +}