diff --git a/docs/spec/core/backend.md b/docs/spec/core/backend.md index 00837adf..b42afdd2 100644 --- a/docs/spec/core/backend.md +++ b/docs/spec/core/backend.md @@ -30,6 +30,7 @@ and react to backend changes. - [Error types](#error-types) - [`LocalBackend` — in-process execution](#localbackend--in-process-execution) - [`RemoteServer` — server-side message handler](#remoteserver--server-side-message-handler) +- [Server-side observability](#server-side-observability) - [`LimitPolicy` — opt-in resource limits](#limitpolicy--opt-in-resource-limits) - [Connection scopes](#connection-scopes) - [`SimulatedRemoteBackend` — adapter for testing](#simulatedremotebackend--adapter-for-testing) @@ -390,6 +391,54 @@ trivially, because every queue is already empty. morph never preempts a running action to force a drain — a model that can run unboundedly long bounds itself; the deadline bounds the *caller's wait*, not the model. +## Server-side observability + +`morph::log` exists and both `RemoteServer` and `QtWebSocketServer` have +access to it, but several outcomes a client cannot distinguish from each +other used to produce no server-side record at all — the operator questions +"did the request arrive? was it rejected? how many clients are connected? +why did that one drop?" had no server-side answer. Four points now log, each +a one-line call at a point the code already reaches: + +- **`RemoteServer::dispatchMessage` — undecodable envelope (`logError`).** A + client that swallows its own error (or is malformed precisely because it + is confused) previously left no trace of a request that never dispatched + at all. Logs the connection id, the exception text, the byte count, and a + **truncated** (256-byte) prefix of the raw payload. The payload is the + most useful field for diagnosing *why* a client sent something malformed — + and the most likely to contain application data, hence truncated rather + than logged in full; server logs should already be treated as + operationally sensitive (they also carry exception text and, on other + lines, connection ids), and this is not a general redaction mechanism. +- **`RemoteServer::dispatchMessage` — one line per successfully-decoded + request (`logDebug`).** `dispatchMessage` is the one place every kind + funnels through, so it is the natural spot: connection id, `kind`, + `callId`, `typeId`/`modelId`/`modelType`/`actionType` (whichever the kind + populates), and the body's byte count. Deliberately omits the session + principal (personal data in many deployments; attribution is still + possible after the fact by correlating `callId`/connection id with the + journal or action log where authentication is configured) and the + request body itself (already covered, truncated, on the decode-failure + path above — logging every successful body by default would be far higher + volume and duplicate what the action log already records for `execute`). + At `debug` because of that volume; a deployment that wants a quieter + default raises `morph::log::setLogLevel` to `info` or higher (the + library's own default minimum level is `debug` — see `logger.hpp` — so + this line *does* appear unless a deployer has already opted into a + quieter stream). +- **`QtWebSocketServer::onNewConnection` — connection refused by + `maxConnections` (`logWarn`).** To the client this looks exactly like the + server being down; the operator previously had no way to learn the cap + was hit, which is the one piece of information that would explain the + symptom. Names the current live count and the configured cap. +- **`QtWebSocketServer::onNewConnection`/`onDisconnected` — connect and + disconnect (`logInfo`).** Neither was recorded before, so there was no way + to reconstruct how many clients were live, or why one went away. Connect + logs the connection id and the live count (including the new connection); + disconnect logs the connection id, the live count (after removal), and the + WebSocket close code + reason — captured from the socket before it is + torn down, since those are the useful part of "why did that one drop?". + ## `SimulatedRemoteBackend` — adapter for testing `SimulatedRemoteBackend` implements `IBackend` by forwarding all calls through @@ -570,6 +619,14 @@ models instead of leaking them: `closeConnection` for every remaining client before aborting its socket, so an orderly server stop also reclaims every client's instances. +**Observability.** `onNewConnection` logs at `morph::log::LogLevel::info` once +a connection is admitted (connection id, live count including the new one); +a connection refused for being over `cfg.maxConnections` logs at `warn` +instead (naming the live count and the configured cap) before the socket is +closed. `onDisconnected` logs at `info` (connection id, live count after +removal, `QWebSocket::closeCode()`/`closeReason()` captured before teardown). +See [Server-side observability](#server-side-observability). + **Flow.** `listen()` binds to the requested TCP port on `cfg.bindAddress` (`QtWebSocketServerConfig`, default `QHostAddress::LocalHost` — today's behavior, unchanged) and starts accepting — unless `cfg.bindAddress` is @@ -634,7 +691,7 @@ reason as `QtWebSocketBackendConfig`) bounds per-connection resource usage: | Field | Default | Enforcement | |---|---|---| -| `maxConnections` | `0` (unbounded) | A connection accepted beyond this count is closed immediately in `onNewConnection`, before any signal is wired or the socket is tracked. | +| `maxConnections` | `0` (unbounded) | A connection accepted beyond this count is closed immediately in `onNewConnection`, before any signal is wired or the socket is tracked. Logged at `morph::log::LogLevel::warn`, naming the live count and the cap — see [Server-side observability](#server-side-observability). | | `maxMessageBytes` | `wire::kMaxEnvelopeBytes` | Checked against the UTF-8 byte length of every incoming frame before it reaches `RemoteServer::handle()`; an oversized frame gets an immediate `err` reply and is never dispatched. The reply carries the rejected call's `callId`, recovered by `wire::detail::peekCallId`'s bounded prefix scan since the frame is deliberately never decoded. A zeroed `callId` would not merely fail to resolve the execute — `0` is the client's synchronous-reply discriminator, so it would resume an unrelated parked `register`/`deregister` with another call's reply. | | `messagesPerSecond` | `0` (unbounded) | A per-connection token bucket (capacity = `messagesPerSecond`, refilled continuously). A frame that finds an empty bucket is dropped silently — not replied to, not queued. | | `handshakeTimeout` | `0` (disabled) | A one-shot timer per connection; if no frame arrives before it fires, the socket is closed. Cancelled on the first frame. Because `QWebSocketServer::newConnection()` only fires after the WS (and TLS, in `SecureMode`) opening handshake completes, this in practice bounds time-to-first-frame after that point, not the handshake itself. | diff --git a/include/morph/core/remote.hpp b/include/morph/core/remote.hpp index 011b67b7..0d443fad 100644 --- a/include/morph/core/remote.hpp +++ b/include/morph/core/remote.hpp @@ -859,9 +859,37 @@ class RemoteServer : public std::enable_shared_from_this { try { env = ::morph::wire::decode(msg); } catch (const std::exception& exc) { + // Undecodable envelope: the one server-side record of a request + // that never dispatched at all. A client that swallows its own + // error (or is malformed precisely because it is confused) would + // otherwise leave no trace here. The payload prefix is the most + // useful field for diagnosing *why* the client sent something + // malformed -- and the most likely to carry application data, so + // it is capped at kLogPayloadPreviewBytes rather than logged in + // full. See docs/spec/core/backend.md, "Server-side observability". + constexpr std::size_t kLogPayloadPreviewBytes = 256; + std::string_view const preview = + std::string_view{msg}.substr(0, std::min(msg.size(), kLogPayloadPreviewBytes)); + ::morph::log::logError("[dispatchMessage] undecodable envelope from connection {}: {} ({} bytes, " + "payload prefix: {}{})", + cid, exc.what(), msg.size(), preview, + msg.size() > kLogPayloadPreviewBytes ? "..." : ""); reply(::morph::wire::encode(::morph::wire::makeErr(exc.what()))); return; } + // One line per successfully-decoded request -- the point every kind + // funnels through, so a client stuck mid-handshake (or one that never + // sent anything) is distinguishable from one whose requests are + // arriving normally. Deliberately omits the session principal (opt-in + // territory: personal data in many deployments) and the payload body + // (already covered, truncated, on the decode-failure path above; + // logging every successful body by default would be far higher volume + // and duplicate what dispatch already records via the action log for + // execute). + ::morph::log::logDebug("[dispatchMessage] connection {}: kind={} callId={} typeId={} modelId={} " + "modelType={} actionType={} bodyBytes={}", + cid, env.kind, env.callId, env.typeId, env.modelId, env.modelType, env.actionType, + env.body.size()); // Once shutdown has begun, new work is rejected fast — before any of // the existing register/execute validation runs — while `deregister` // (and any other kind) still flows through unchanged, so a client can diff --git a/src/qt/qt_websocket_server.cpp b/src/qt/qt_websocket_server.cpp index 08bd07ab..ea30b897 100644 --- a/src/qt/qt_websocket_server.cpp +++ b/src/qt/qt_websocket_server.cpp @@ -139,6 +139,11 @@ void QtWebSocketServer::onNewConnection() { return; } if (_cfg.maxConnections != 0 && _clients.size() >= _cfg.maxConnections) { + // To the client this looks exactly like the server being down -- the + // one piece of information that would actually explain the symptom + // (the configured cap was hit) otherwise never reaches an operator. + ::morph::log::logWarn("[QtWebSocketServer] connection refused: at maxConnections ({}/{})", _clients.size(), + _cfg.maxConnections); socket->close(); socket->deleteLater(); return; @@ -153,6 +158,9 @@ void QtWebSocketServer::onNewConnection() { state.lastActivity = state.lastRefill; state.cid = _server.openConnection(); + // +1: this connection is admitted but not yet in `_clients` at this point. + ::morph::log::logInfo("[QtWebSocketServer] connection {} accepted ({} live)", state.cid, _clients.size() + 1); + if (_cfg.handshakeTimeout.count() > 0) { auto* timer = new QTimer(this); timer->setSingleShot(true); @@ -248,6 +256,13 @@ void QtWebSocketServer::onDisconnected() { } auto iter = _clients.find(socket); if (iter != _clients.end()) { + // Close code and reason are only available on the socket while it is + // still around; capture them before erase()/deleteLater() below -- + // they are the useful part of "why did that one drop?", the question + // an operator otherwise has no way to answer from the server side. + ::morph::log::logInfo("[QtWebSocketServer] connection {} disconnected ({} live), closeCode={} reason={}", + iter->second.cid, _clients.size() - 1, static_cast(socket->closeCode()), + socket->closeReason().toStdString()); _server.closeConnection(iter->second.cid); if (iter->second.handshakeTimer) { iter->second.handshakeTimer->stop(); diff --git a/tests/qt/test_qt_websocket.cpp b/tests/qt/test_qt_websocket.cpp index 951083b7..1ae31c33 100644 --- a/tests/qt/test_qt_websocket.cpp +++ b/tests/qt/test_qt_websocket.cpp @@ -18,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -28,6 +29,8 @@ #include #include #include +#include +#include // ── Shared QCoreApplication ────────────────────────────────────────────────── // QCoreApplication is owned by main() (below) and torn down before any global @@ -763,6 +766,86 @@ TEST_CASE("morph::qt::QtWebSocketServer: maxConnections rejects connections beyo pumpUntil([&] { return first.state() == QAbstractSocket::UnconnectedState; }, 50); } +TEST_CASE("morph::qt::QtWebSocketServer: maxConnections refusal is logged at warn, naming the cap", + "[qt][ws][limits][issue30]") { + ensureApp(); + morph::exec::ThreadPoolExecutor serverPool{2}; + auto server = std::make_shared(serverPool); + morph::qt::QtWebSocketServerConfig cfg; + cfg.maxConnections = 1; + morph::qt::QtWebSocketServer wsServer{*server, 0, std::nullopt, cfg}; + REQUIRE(wsServer.listen()); + + QUrl url{QString("ws://127.0.0.1:%1").arg(wsServer.port())}; + + QWebSocket first; + first.open(url); + pumpUntil([&] { return first.state() == QAbstractSocket::ConnectedState; }, 100); + REQUIRE(first.state() == QAbstractSocket::ConnectedState); + + std::vector> captured; + { + morph::log::ScopedLoggerOverride guard{ + [&](morph::log::LogLevel level, std::string_view msg) { captured.emplace_back(level, std::string{msg}); }, + morph::log::LogLevel::debug}; + + QWebSocket second; + second.open(url); + pumpUntil([&] { return second.state() == QAbstractSocket::UnconnectedState; }, 150); + REQUIRE(second.state() == QAbstractSocket::UnconnectedState); + } + + bool foundWarning = false; + for (auto const& [level, msg] : captured) { + if (level == morph::log::LogLevel::warn && msg.find("maxConnections") != std::string::npos) { + foundWarning = true; + break; + } + } + CHECK(foundWarning); + + first.close(); + pumpUntil([&] { return first.state() == QAbstractSocket::UnconnectedState; }, 50); +} + +TEST_CASE("morph::qt::QtWebSocketServer: connect/disconnect are logged at info, with the live count", + "[qt][ws][issue30]") { + ensureApp(); + morph::exec::ThreadPoolExecutor serverPool{2}; + auto server = std::make_shared(serverPool); + morph::qt::QtWebSocketServer wsServer{*server, 0}; + REQUIRE(wsServer.listen()); + QUrl url{QString("ws://127.0.0.1:%1").arg(wsServer.port())}; + + std::vector> captured; + { + morph::log::ScopedLoggerOverride guard{ + [&](morph::log::LogLevel level, std::string_view msg) { captured.emplace_back(level, std::string{msg}); }, + morph::log::LogLevel::debug}; + + QWebSocket client; + client.open(url); + pumpUntil([&] { return client.state() == QAbstractSocket::ConnectedState; }, 100); + REQUIRE(client.state() == QAbstractSocket::ConnectedState); + + client.close(); + pumpUntil([&] { return client.state() == QAbstractSocket::UnconnectedState; }, 100); + } + + bool foundConnect = false; + bool foundDisconnect = false; + for (auto const& [level, msg] : captured) { + if (level == morph::log::LogLevel::info && msg.find("accepted") != std::string::npos) { + foundConnect = true; + } + if (level == morph::log::LogLevel::info && msg.find("disconnected") != std::string::npos) { + foundDisconnect = true; + } + } + CHECK(foundConnect); + CHECK(foundDisconnect); +} + TEST_CASE("morph::qt::QtWebSocketServer: maxMessageBytes rejects an oversized frame before dispatch", "[qt][ws][limits]") { ensureApp(); diff --git a/tests/qt/test_qt_websocket_adversarial.cpp b/tests/qt/test_qt_websocket_adversarial.cpp index 90c1371c..7a89fd20 100644 --- a/tests/qt/test_qt_websocket_adversarial.cpp +++ b/tests/qt/test_qt_websocket_adversarial.cpp @@ -40,9 +40,12 @@ #include #include #include +#include #include #include #include +#include +#include #include #include #include @@ -221,6 +224,57 @@ TEST_CASE("Adversarial: duplicate-JSON-key envelope does not crash the server", requireServerStillServesHonestClients(url); } +TEST_CASE("Adversarial: an undecodable envelope is logged at error, with a truncated payload preview, and does " + "not take down the server", + "[qt][ws][adversarial][issue30]") { + morph::exec::ThreadPoolExecutor serverPool{2}; + auto server = std::make_shared(serverPool); + morph::qt::QtWebSocketServer wsServer{*server, 0}; + REQUIRE(wsServer.listen()); + const QUrl url{QString("ws://127.0.0.1:%1").arg(wsServer.port())}; + + QWebSocket hostile; + hostile.open(url); + pumpUntil([&] { return hostile.state() == QAbstractSocket::ConnectedState; }, 100); + REQUIRE(hostile.state() == QAbstractSocket::ConnectedState); + + QString reply; + bool got = false; + auto conn = QObject::connect(&hostile, &QWebSocket::textMessageReceived, [&](const QString& msg) { + reply = msg; + got = true; + }); + + std::vector> captured; + static const QString kGarbage = QStringLiteral("this is not JSON at all { [ malformed morph_probe_marker"); + { + morph::log::ScopedLoggerOverride guard{ + [&](morph::log::LogLevel level, std::string_view msg) { captured.emplace_back(level, std::string{msg}); }, + morph::log::LogLevel::debug}; + hostile.sendTextMessage(kGarbage); + pumpUntil([&] { return got; }, 100); + } + QObject::disconnect(conn); + REQUIRE(got); + // Rejected with an err reply -- not silently dropped, and not a crash. + auto decodedReply = morph::wire::decode(reply.toStdString()); + CHECK(decodedReply.kind == "err"); + + bool foundLogLine = false; + for (auto const& [level, msg] : captured) { + if (level == morph::log::LogLevel::error && msg.find("dispatchMessage") != std::string::npos && + msg.find("morph_probe_marker") != std::string::npos) { + foundLogLine = true; + break; + } + } + CHECK(foundLogLine); + + hostile.close(); + pumpUntil([&] { return hostile.state() == QAbstractSocket::UnconnectedState; }, 50); + requireServerStillServesHonestClients(url); +} + TEST_CASE("Adversarial: a client that opens then stalls does not block other clients", "[qt][ws][adversarial]") { morph::exec::ThreadPoolExecutor serverPool{2}; auto server = std::make_shared(serverPool);