diff --git a/docs/spec/core/backend.md b/docs/spec/core/backend.md index 01cc9c4f8..9c625a909 100644 --- a/docs/spec/core/backend.md +++ b/docs/spec/core/backend.md @@ -43,6 +43,7 @@ and react to backend changes. - [`QtWebSocketBackend` — client-side WebSocket transport](#qtwebsocketbackend--client-side-websocket-transport) - [`QtWebSocketServer` — server-side WebSocket transport](#qtwebsocketserver--server-side-websocket-transport) - [`SocketBackend` / `SocketServer` — raw-socket WebSocket transport](#socketbackend--socketserver--raw-socket-websocket-transport) + - [The structural registration surface, natively](#the-structural-registration-surface-natively) - [Lifetime & ownership](#lifetime--ownership) - [Failure modes](#failure-modes) - [Thread context](#thread-context) @@ -410,11 +411,13 @@ morph::backend::SynchronousBackendAdapter adapter{local, pool}; It is what makes morph#522's steps 2–5 migrations rather than rewrites: every backend that stays blocking — `LocalBackend`, `SimulatedRemoteBackend`, and -eleven test doubles, plus `SocketBackend` if morph#569 decides a wrapper is -right for it — reaches the new surface without being edited. The one backend -that has a real non-blocking path, `QtWebSocketBackend`, deliberately does not -use this: morph#568 moves it across natively, because wrapping it would -reintroduce the blocking this exists to route around. +eleven test doubles — reaches the new surface without being edited. Two +backends deliberately do not use it, for the same reason in two different +shapes: `QtWebSocketBackend` (morph#568) and `SocketBackend` (morph#569) both +have a genuinely non-blocking path of their own, and wrapping either would +reintroduce the blocking this exists to route around. `SocketBackend`'s case is +set out under [The structural registration surface, +natively](#the-structural-registration-surface-natively). - **It does not make blocking non-blocking.** The wrapped backend still blocks. What changes is which thread pays: the blocking call runs on the adapter's @@ -430,11 +433,17 @@ reintroduce the blocking this exists to route around. the executor must still be running tasks when the adapter is destroyed — the same rule as `StrandExecutor`'s own `base`. - **A control call issued from a reconnect handler runs on the strand**, never - on the wrapped backend's transport thread. Whether that settles - `SocketBackend`'s documented reconnect-handler deadlock hazard (see the - "runs reconnect handlers on a dedicated thread" row under [Design - decisions](#design-decisions)) is morph#569's question; this page does not - claim it does. + on the wrapped backend's transport thread — *provided the handler issues it + through `bindModel`/`promoteModel`*. That proviso is load-bearing, and it is + why the adapter does **not** settle `SocketBackend`'s documented + reconnect-handler deadlock hazard (see the "runs reconnect handlers on a + dedicated thread" row under [Design decisions](#design-decisions)): the + adapter forwards `setReconnectHandler` to the wrapped backend unchanged, and + `Bridge::installReconnectHandler`'s handler calls the *blocking* + `registerModelShared`/`registerModelWithContext`, which the adapter also + forwards unchanged. A wrapped `SocketBackend` would therefore run its + reconnect control calls exactly where it runs them today. See morph#569's + answer under [`SocketBackend`](#socketbackend--socketserver--raw-socket-websocket-transport). ### Backends with a genuinely non-blocking path @@ -443,17 +452,19 @@ the `Completion` when its reply arrives. There is no `bool`, no fallback verb and no inline-completion special case to declare: settling the `Completion` inside the dispatch call and settling it a second later from a transport thread are the same code at the call site, because delivery goes through `cbExec` -either way. That is the shape morph#568 moves `QtWebSocketBackend` onto. +either way. That is the shape morph#568 moves `QtWebSocketBackend` onto, and +the shape morph#569 moves `SocketBackend` onto. ### Migration status -Nothing in the tree uses this surface yet. `Bridge` still calls the five -synchronous verbs and prefers the four `*Async` twins, every existing -implementor still compiles unchanged, and the default `bindModel`/`promoteModel` -implementations route to exactly the legacy verb each request shape names — so a -backend that has overridden nothing behaves identically through either surface. -The four twins are removed, and the prose threading contract retired, in -morph#571. +No *caller* uses this surface yet. `Bridge` still calls the five synchronous +verbs and prefers the four `*Async` twins, every existing implementor still +compiles unchanged, and the default `bindModel`/`promoteModel` implementations +route to exactly the legacy verb each request shape names — so a backend that +has overridden nothing behaves identically through either surface. On the +implementor side, `SocketBackend` overrides both natively (morph#569) without +changing any of its legacy verbs. The four twins are removed, and the prose +threading contract retired, in morph#571. ## Error types @@ -1501,6 +1512,76 @@ throws is caught and logged, so the next reconnect still finds the thread waiting. `QtWebSocketBackend` has no equivalent need — its `sendSync` runs a nested `QEventLoop` that keeps pumping the socket. +### The structural registration surface, natively + +`SocketBackend` overrides `bindModel`/`promoteModel` itself rather than being +wrapped in [`SynchronousBackendAdapter`](#synchronousbackendadapter--a-blocking-backend-on-the-new-surface). +It had overridden none of the four `*Async` verbs, so the wrapper was the +expected route; three findings decided against it (morph#569). + +1. **The transport already has the machinery.** The I/O thread demultiplexes + replies by `callId` for `execute`, and `RemoteServer` echoes `callId` on + every control reply it sends — `register`, `registerShared`, `attach` and + `assign` all answer with `makeOk(env.callId, {}, mid)`. A control call is + therefore the same shape as an execute, and the native path needs no + protocol change, no server change and no new thread. It is a second + `PendingCallTable`, sharing `_pending`'s `callId` counter so an id can never + be ambiguous between the two. +2. **A wrapper would keep every bind inside `sendSync`'s one-call token.** + `sendSync` admits exactly one synchronous control call across the whole + backend and throws `"a synchronous call is already in flight (reentrant + use)"` on a second. The adapter's strand serialises binds against *each + other*, but `listInstances` and the legacy `registerModel` are not on that + strand, and this backend is explicitly documented as safe to drive from + several threads at once. The native path takes no token at all. +3. **The adapter's reconnect property does not apply here** — see the answer + below, which is the question morph#569 was opened to settle. + +What does **not** change: `registerModel`, `registerModelShared`, `attachModel` +and `assignPrimary` still use `sendSync` with `callId == 0`, so every caller +morph#570/#571 has yet to migrate behaves exactly as before, and nothing on the +wire changes for them. `cancelPending` now sweeps both tables, so a disconnect +rejects an in-flight bind with `DisconnectedError` instead of stranding it — +the asynchronous counterpart of `sendSync` waking on `!_connected`. + +**What the new surface does to the reconnect-handler deadlock hazard.** The +hazard is that a control call parks on `_syncCv` waiting for a reply only the +I/O thread's read loop can deliver, so running one *on* that thread blocks the +thread that would satisfy it. Stated precisely, in three parts: + +- **Through the adapter it would be untouched.** Not relocated — untouched. The + adapter forwards `setReconnectHandler` to the wrapped backend, so the handler + still runs wherever `SocketBackend` chooses to run it, and + `Bridge::installReconnectHandler`'s handler calls the blocking verbs, which + the adapter also forwards unchanged. Nothing about that path would have gone + near the strand. +- **Natively, a bind cannot have the hazard at all.** `bindModel` never enters + `sendSync`, never waits on `_syncCv`, and returns before the reply exists, so + there is no wait for any thread to block — including the I/O thread itself. + That is structural, not a mitigation: it follows from the signature returning + a `Completion` rather than a `ModelId`, and it holds whichever thread issues + the call. +- **The hazard is nevertheless still present in the backend, and the dedicated + handler thread is still load-bearing.** The blocking verbs still park on + `_syncCv`, and `Bridge`'s reconnect handler still calls them. A hazard-free + route now exists; the hazardous one has not been removed or moved. Only when + morph#570 puts `Bridge::installReconnectHandler` on `bindModel` does the + hazard become unreachable from the reconnect path, and only then is dropping + the handler thread a question worth asking. + +None of this touches morph#486. The surface adds no lock and knows nothing +about a caller's teardown; it moves the choice of delivery thread from the +implementor to the caller, exactly as +[How the threading contract becomes structural](#how-the-threading-contract-becomes-structural) +says and no further. + +`tests/net/test_socket_backend.cpp` pins the property the argument rests on +rather than only its result: a bind is accepted and settles *while a legacy +synchronous call is still parked on `_syncCv`* — which a blocking bind cannot +be, because it fails there with the reentrant-use error — and four binds are in +flight simultaneously, matched by `callId`, with the replies delivered back to +front. + **Threading — the one deliberate difference from the Qt transport.** `QtWebSocketBackend` is pinned to the Qt event loop and uses a nested `QEventLoop` for its synchronous `registerModel`; `SocketBackend` instead owns @@ -1785,7 +1866,10 @@ thread instead of the Qt thread: Unlike `QtWebSocketBackend`, `SocketBackend`'s `execute`/`registerModel`/ `deregisterModel` may themselves be called from any thread — there is no -single owning event-loop thread to violate. `morph::net::SocketServer` +single owning event-loop thread to violate. `bindModel`/`promoteModel` are +likewise callable from any thread, including the I/O thread itself: they park +on nothing, and their continuations run on the caller's `cbExec` rather than on +the I/O thread that settles them. `morph::net::SocketServer` receives frames on its own per-connection thread, hands them to `RemoteServer::handle` (server pool / model strand, as above), and writes the reply back on whichever thread produces it (serialized per connection by a @@ -1979,11 +2063,13 @@ not a behavior change to the existing loopback-only default. | `explicit SocketBackend(serverUrl, cfg = Config{})` | Parses `serverUrl` (`ws://` only — throws immediately on `wss://`) and starts the I/O thread, which connects asynchronously. | | `waitForConnected(timeout = 5000ms)` | Blocks the calling thread on a condition variable until connected or the timeout elapses; returns the current connected state. The backend must outlive the call — destroying it while a thread is parked here is undefined, and there is no cancel (see Lifetime & ownership). | | `registerModel(typeId, factory)` | Synchronous via a parked condition variable; `factory` ignored. Throws on `err` reply or disconnect. Thread-safe, but only one such call may be in flight at a time. | +| `bindModel(request, cbExec)` | Native override of the structural surface. Sends the envelope `request`'s shape names with a non-zero `callId` and returns immediately; the I/O thread settles the `Completion` when the reply arrives, delivered on `cbExec`. Never enters `sendSync`, so it takes no synchronous-call token and any number may be in flight. Rejects with `DisconnectedError` when the socket is down or drops first, or with `std::runtime_error{" failed: "}`. | +| `promoteModel(request, cbExec)` | The `assign` counterpart of `bindModel`, on the same path; resolves with `request.mid` echoed back. An empty `primary` or a zero `mid` resolves without sending, matching `assignPrimary`'s guards. | | `deregisterModel(mid)` | **Fire-and-forget** — sends only if connected, does not wait for the ack. Carries a non-zero `callId` from the same counter `execute` uses so its unawaited `ok` cannot be handed to a parked synchronous control call (issue #454; the `QtWebSocketBackend` precedent is issue #65). Needs no pending-id bookkeeping of its own: `dispatchIncomingEnvelope` already drops a non-zero `callId` that is absent from `_pending`. | | `execute(mid, call, cbExec)` | Assigns a `callId`, sends `execute`, returns a `Completion`. Immediate `DisconnectedError` if not connected. Thread-safe; supports concurrent in-flight calls from multiple threads. | | `notifyBackendChanged()` | No-op. | -| `cancelPending(exc)` | Drains the pending map, delivers `exc` to each state. | -| `setReconnectHandler(handler)` | Stores the handler; invoked on the I/O thread after every *subsequent* connect. `nullptr` clears. | +| `cancelPending(exc)` | Drains **both** pending maps — the `execute` calls and the `bindModel`/`promoteModel` control calls — and delivers `exc` to each state. | +| `setReconnectHandler(handler)` | Stores the handler; invoked on the **dedicated handler thread**, not the I/O thread, after every *subsequent* connect — see "Reconnect handlers run on their own thread" above. `nullptr` clears. | ### `SocketServerConfig` (`morph::net::SocketServer::Config`) @@ -2034,7 +2120,8 @@ not a behavior change to the existing loopback-only default. | One `bindModel` instead of three acquire verbs | Behaviour selected by `BindRequest`'s shape (`primary` empty?, `current` zero?) | The three verbs already degrade into each other exactly along those two fields, so naming them separately stated the same distinction three times — and tripled it again for the `*Async` twins. The default implementation still routes each shape to the legacy verb it names, so a backend that overrides only some of the three is unaffected. | | `SynchronousBackendAdapter` is a decorator, not a base class or a CRTP mixin | Wraps `shared_ptr` and forwards every verb | It must work on `LocalBackend` and eleven test doubles *without modifying them*, which rules out anything they would have to derive from. Cost is one forwarding method per unchanged verb; benefit is that morph#522's remaining four steps are migrations rather than rewrites. | | The adapter's blocking executor is required, not defaulted | Constructor parameter with no default; null `inner` throws | "Where does the blocking happen" is the only question the class exists to answer. An adapter that silently ran the call inline when handed nothing would block on some configurations and not others — contract by configuration, which is the thing being removed. | -| `SocketBackend` runs reconnect handlers on a dedicated thread | Not inline from the I/O thread's connect path | A reconnect handler re-registers models via the synchronous control path, which waits for a reply only the I/O thread's read loop can deliver. Inline, that wait blocks the very thread that would satisfy it, deadlocking the transport with no timeout. | +| `SocketBackend` runs reconnect handlers on a dedicated thread | Not inline from the I/O thread's connect path | A reconnect handler re-registers models via the synchronous control path, which waits for a reply only the I/O thread's read loop can deliver. Inline, that wait blocks the very thread that would satisfy it, deadlocking the transport with no timeout. Still load-bearing after morph#569: `bindModel` cannot deadlock this way, but the blocking verbs still can and `Bridge`'s reconnect handler still calls them. | +| `SocketBackend` implements `bindModel`/`promoteModel` natively | Not wrapped in `SynchronousBackendAdapter`, although it overrides none of the four `*Async` verbs | The I/O thread already demultiplexes replies by `callId` for `execute` and `RemoteServer` echoes `callId` on every control reply, so the non-blocking path costs a second `PendingCallTable` and no protocol change. Wrapping instead would park a thread per bind for a round trip this transport need not park for, and would keep every bind inside `sendSync`'s one-synchronous-call token — which `listInstances` and the legacy `registerModel` share, and which is the one global restriction on a backend otherwise documented as drivable from several threads at once. The adapter's reconnect-handler property, the other reason to consider it, does not apply: it forwards `setReconnectHandler` and the blocking verbs straight through, so a wrapped `SocketBackend` would run reconnect control calls exactly where it does today. | ## Lifetime annotations diff --git a/include/morph/net/socket_backend.hpp b/include/morph/net/socket_backend.hpp index 73ce46593..c8c1d8f25 100644 --- a/include/morph/net/socket_backend.hpp +++ b/include/morph/net/socket_backend.hpp @@ -261,6 +261,97 @@ class SocketBackend : public ::morph::backend::detail::IBackend { (void)sendControlForId(env, "assign"); } + // ── The structural registration surface (morph#567 / morph#569) ────── + // + // Overridden natively rather than reached through + // `backend::SynchronousBackendAdapter`. The reasoning is recorded in + // docs/spec/core/backend.md (`SocketBackend`'s section, "The structural + // registration surface, natively"); the short form is that this transport + // already demultiplexes replies by `callId` on its I/O thread for + // `execute`, and a control call is the same shape. Running the *blocking* + // verb on a wrapper's strand would park a thread for a round trip this + // transport need not park for, and would keep every bind inside + // `sendSync`'s one-synchronous-call-at-a-time token — which a concurrent + // `listInstances` or legacy `registerModel` on another thread shares, and + // which this backend is otherwise documented as not having (it may be + // driven from several threads at once). + // + // Nothing about the legacy verbs changes: `registerModel`, + // `registerModelShared`, `attachModel` and `assignPrimary` still use + // `sendSync` and `callId == 0`, so every caller morph#570/#571 has yet to + // migrate behaves exactly as before. + + /// @brief Acquires a model instance without blocking the calling thread. + /// + /// Sends the control envelope @p request's shape names (see + /// `backend::detail::BindRequest`'s table) carrying a non-zero `callId` + /// drawn from the same counter `execute` uses, and settles the returned + /// `Completion` from the I/O thread when the matching reply arrives. No + /// thread is parked anywhere: in particular this never enters `sendSync`, + /// so a bind neither waits on `_syncCv` nor takes the one-synchronous-call + /// token, and therefore cannot wait on the I/O thread that would satisfy it. + /// + /// The two degradations the legacy verbs perform are preserved exactly: an + /// empty `primary` with a live `current` gives that instance up first, and + /// an empty `primary` binds a private instance. + /// + /// @param request Owning bind request; moved from. + /// @param cbExec Executor the continuation is delivered on. Borrowed: it + /// must outlive the returned `Completion`. + /// @return A `Completion` resolved with the bound `ModelId`; rejected with + /// `backend::DisconnectedError` if the socket is down or drops + /// before the reply, or with a `std::runtime_error` carrying the + /// server's own error message. + ::morph::async::Completion<::morph::exec::detail::ModelId> bindModel(::morph::backend::detail::BindRequest request, + ::morph::exec::IExecutor& cbExec) override { + if (request.primary.empty() && request.current.v != 0U) { + // `attachModel`'s empty-primary branch: the instance being given up + // is released before the private bind that replaces it. + deregisterModel(request.current); + } + if (request.primary.empty()) { + // `contextKey` is dropped here because the blocking path drops it: + // `IBackend::registerModelWithContext`'s default forwards to + // `registerModel` and discards it, and this backend does not + // override it. Keeping the native path bit-for-bit identical + // matters more than changing that here; it is filed separately. + return sendControlAsync(::morph::wire::makeRegister(request.typeId), "register", std::nullopt, cbExec); + } + if (request.current.v != 0U) { + return sendControlAsync( + ::morph::wire::makeAttach(request.typeId, request.primary, request.current.v, request.contextKey), + "attach", std::nullopt, cbExec); + } + return sendControlAsync(::morph::wire::makeRegisterShared(request.typeId, request.primary, request.contextKey), + "register", std::nullopt, cbExec); + } + + /// @brief Files an already-live server-side instance under a key, without + /// blocking the calling thread. + /// + /// The promote counterpart of `bindModel`, on the same callId-multiplexed + /// path. Resolves with `request.mid` echoed back exactly as + /// `IBackend::promoteModel` documents — including for the two guards + /// `assignPrimary` applies locally (empty `primary`, zero `mid`), which + /// resolve without sending anything. + /// + /// @param request Owning promote request; moved from. + /// @param cbExec Executor the continuation is delivered on. Borrowed: it + /// must outlive the returned `Completion`. + /// @return A `Completion` resolved with `request.mid`, or rejected as + /// `bindModel` documents. + ::morph::async::Completion<::morph::exec::detail::ModelId> promoteModel( + ::morph::backend::detail::PromoteRequest request, ::morph::exec::IExecutor& cbExec) override { + if (request.primary.empty() || request.mid.v == 0U) { + auto state = std::make_shared<::morph::async::detail::CompletionState<::morph::exec::detail::ModelId>>(); + ::morph::async::Completion<::morph::exec::detail::ModelId> comp{state, &cbExec}; + state->setValue(request.mid); + return comp; + } + return sendControlAsync(::morph::wire::makeAssign(request.typeId, request.primary, request.mid.v), "assign", + request.mid, cbExec); + } + /// @brief Asks the server for the live shared primary keys of @p typeId. /// @param typeId String type-id to enumerate. /// @return Canonical key strings of the live shared instances. @@ -389,7 +480,13 @@ class SocketBackend : public ::morph::backend::detail::IBackend { /// @brief No-op — this backend holds no local model objects. void notifyBackendChanged() override {} - /// @brief Resolves every pending execute call's `Completion` with @p exc. + /// @brief Resolves every pending call's `Completion` with @p exc. + /// + /// Covers both in-flight tables: the `execute` calls in `_pending` and the + /// `bindModel`/`promoteModel` control calls in `_pendingControl`. A bind + /// left out of this sweep would hang forever on a disconnect, since its + /// reply can now only arrive on a connection that is gone — the async + /// counterpart of `sendSync`'s `!_connected` wake-up. /// @param exc Exception delivered to every pending completion's error sink. void cancelPending(const std::exception_ptr& exc) override { auto drained = _pending.drain(); @@ -399,10 +496,19 @@ class SocketBackend : public ::morph::backend::detail::IBackend { pending.state->setException(exc); } } + auto drainedControl = _pendingControl.drain(); + for (auto& [callId, pending] : drainedControl) { + (void)callId; + if (pending.state) { + pending.state->setException(exc); + } + } } /// @brief Installs the handler invoked after each *subsequent* successful (re)connect. - /// @param handler Callable invoked on the I/O thread. Pass `nullptr` to clear. + /// @param handler Callable invoked on this backend's dedicated handler + /// thread — deliberately not the I/O thread, see + /// `onConnected`. Pass `nullptr` to clear. void setReconnectHandler(const std::function& handler) override { std::scoped_lock lock{_reconnectHandlerMtx}; _reconnectHandler = handler; @@ -430,6 +536,89 @@ class SocketBackend : public ::morph::backend::detail::IBackend { ::morph::exec::IExecutor* cbExec{nullptr}; }; + /// @brief One in-flight control call issued through `bindModel`/`promoteModel`. + /// + /// Kept in its own table rather than in `_pending`: an execute reply + /// settles a `Completion>` through a deserializer, a + /// control reply settles a `Completion` and has none. The two + /// tables share one `callId` counter (`_pending.nextCallId()`), so an id is + /// never ambiguous between them. + struct PendingControl { + /// @brief State of the `Completion` this call settles. + std::shared_ptr<::morph::async::detail::CompletionState<::morph::exec::detail::ModelId>> state; + /// @brief Verb name prefixing the server's error message, matching the + /// legacy verbs' `" failed: ..."` wording. + std::string what; + /// @brief Id to resolve with, for `promoteModel`, which echoes its + /// request's `mid`. `nullopt` means "resolve with the reply's". + std::optional<::morph::exec::detail::ModelId> echo; + }; + + /// @brief Sends one control envelope on the callId-multiplexed path and + /// returns the `Completion` its reply will settle. + /// @param env Envelope to send; its `callId` and `session` are filled in here. + /// @param what Verb name for the error message. + /// @param echo Id to resolve with, or `nullopt` to use the reply's `modelId`. + /// @param cbExec Executor the continuation is delivered on. + /// @return The `Completion` the reply — or a disconnect — settles. + ::morph::async::Completion<::morph::exec::detail::ModelId> sendControlAsync( + ::morph::wire::Envelope env, std::string_view what, std::optional<::morph::exec::detail::ModelId> echo, + ::morph::exec::IExecutor& cbExec) { + auto state = std::make_shared<::morph::async::detail::CompletionState<::morph::exec::detail::ModelId>>(); + ::morph::async::Completion<::morph::exec::detail::ModelId> comp{state, &cbExec}; + + std::uint64_t const callId = _pending.nextCallId(); + env.callId = callId; + env.session = currentSession(); + std::string payload; + try { + payload = ::morph::wire::encode(env); + } catch (const std::exception& exc) { + state->setException( + std::make_exception_ptr(std::runtime_error(std::string{what} + " failed: " + exc.what()))); + return comp; + } + + // Admitted under the table's own lock, for the reason `execute` spells + // out: a disconnect sweep (`cancelPending` -> `drain()`) running between + // a bare `_connected` check and the insert would file this entry after + // the sweep already emptied the table, and nothing would settle it. + if (!_pendingControl.insertIf(callId, PendingControl{.state = state, .what = std::string{what}, .echo = echo}, + [this] { return _connected.load(); })) { + state->setException(std::make_exception_ptr(::morph::backend::DisconnectedError{})); + return comp; + } + + try { + sendFrame(::morph::net::detail::WsOpcode::kText, payload); + } catch (const std::exception&) { + // The write either raced a disconnect already under way or (as + // `sendFrame` documents) tore the connection down itself. Reclaim + // the entry rather than leave it to the io thread's sweep: `take` + // is atomic, so if the sweep won the race it already settled this + // state and there is nothing here to settle. + if (auto reclaimed = _pendingControl.take(callId)) { + state->setException(std::make_exception_ptr(::morph::backend::DisconnectedError{})); + } + } + return comp; + } + + /// @brief Settles one control call from its matched reply envelope. + /// @param pending Entry taken out of `_pendingControl`. + /// @param reply Decoded reply carrying the same `callId`. + static void settleControl(const PendingControl& pending, const ::morph::wire::Envelope& reply) { + if (!pending.state) { + return; + } + if (reply.kind == "ok") { + pending.state->setValue(pending.echo.value_or(::morph::exec::detail::ModelId{reply.modelId})); + return; + } + pending.state->setException( + std::make_exception_ptr(std::runtime_error(pending.what + " failed: " + reply.message))); + } + void sendFrame(::morph::net::detail::WsOpcode opcode, std::string_view payload) { std::scoped_lock lock{_socketMtx}; if (!_socket.valid()) { @@ -591,6 +780,15 @@ class SocketBackend : public ::morph::backend::detail::IBackend { if (env.callId != 0U) { auto pending = _pending.take(env.callId); if (!pending) { + // Not an execute: it may be a control reply for a `bindModel`/ + // `promoteModel` in flight, which shares this id space. Settling + // it here — on the io thread, before `readLoop` asks for the + // next frame — is what lets a control call be issued without + // parking any thread on `_syncCv`. + if (auto control = _pendingControl.take(env.callId)) { + settleControl(*control, env); + return; + } return; // late/cancelled reply — dropped silently } // Triage shared with SimulatedRemoteBackend and QtWebSocketBackend @@ -782,6 +980,10 @@ class SocketBackend : public ::morph::backend::detail::IBackend { std::optional _syncReply; ::morph::backend::detail::PendingCallTable _pending; + // Control calls issued through the structural surface. Deliberately shares + // `_pending`'s callId counter (allocated via `_pending.nextCallId()`) rather + // than owning a second one, so the two tables can never claim the same id. + ::morph::backend::detail::PendingCallTable _pendingControl; std::mutex _reconnectHandlerMtx; std::function _reconnectHandler; diff --git a/scripts/branch_partial_allowlist.json b/scripts/branch_partial_allowlist.json index 82d19b6d9..bb8c9734f 100644 --- a/scripts/branch_partial_allowlist.json +++ b/scripts/branch_partial_allowlist.json @@ -146,7 +146,7 @@ }, { "file": "include/morph/net/socket_backend.hpp", - "line": 615, + "line": 813, "source": "default:", "reason": "Unreachable except via the adjacent `Error` case it deliberately shares a body with (net audit, `socket_backend.hpp` extra finding #7). `detail::ExecuteReplyKind` is a closed 3-value enum (`Value`/`Timeout`/`Error`), all three handled explicitly above this label; `default:` exists only to satisfy this project's `-Wswitch-default`, per the source's own inline comment directly below this line. Reaching it via any value other than through the `Error` case falling through would require an out-of-range `static_cast` producing a value outside the enum's domain -- undefined behavior, not a legitimate test target." } diff --git a/tests/net/test_socket_backend.cpp b/tests/net/test_socket_backend.cpp index 983dd6496..aacd57497 100644 --- a/tests/net/test_socket_backend.cpp +++ b/tests/net/test_socket_backend.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -1687,3 +1688,439 @@ TEST_CASE("SocketBackend: many concurrent executes racing a disconnect leave no std::this_thread::sleep_for(std::chrono::milliseconds{50}); } } + +// ── The structural registration surface (morph#569) ────────────────────────── +// +// `SocketBackend` overrides `bindModel`/`promoteModel` natively rather than +// being wrapped in `SynchronousBackendAdapter`. The property that decides that +// choice, and that the deadlock argument in docs/spec/core/backend.md rests on, +// is that a native control call never enters `sendSync`: it goes out on the +// same callId-multiplexed path `execute` uses and is settled by the I/O +// thread's read loop. The tests below pin that property, not just the +// functional result. + +namespace { + +// Drains `exec` on the calling thread until `done`, or gives up. The caller's +// executor is a MainThreadExecutor precisely so that "the continuation ran" +// and "the caller pumped it" are separable events. +bool drainUntil(morph::exec::MainThreadExecutor& exec, const std::atomic& done, int maxIterations = 500) { + for (int i = 0; i < maxIterations && !done.load(); ++i) { + if (!exec.runOnce()) { + std::this_thread::sleep_for(std::chrono::milliseconds{5}); + } + } + return done.load(); +} + +morph::backend::detail::BindRequest privateBind(std::string typeId) { + return morph::backend::detail::BindRequest{ + .typeId = std::move(typeId), .factory = nullptr, .contextKey = {}, .primary = {}, .current = {}}; +} + +} // namespace + +TEST_CASE( + "SocketBackend: bindModel reaches the server for each BindRequest shape and settles on the caller's executor", + "[net][socket_backend][registration-surface]") { + morph::exec::ThreadPoolExecutor serverPool{2}; + auto server = std::make_shared(serverPool); + morph::net::SocketServer wsServer{*server, 0}; + REQUIRE(wsServer.listen()); + + // Declared before `backend`, so it is destroyed *after* it: `~SocketBackend` + // joins the I/O thread, which can call `post()` on this executor right up + // until that join completes. With the reverse order, TSan caught the I/O + // thread still running -- and still able to call `post()` -- while this + // executor's own destructor was tearing down its condition variable on the + // main thread (morph#586, data race in pthread_cond_destroy). + morph::exec::MainThreadExecutor callerExec; + morph::net::SocketBackend backend{"ws://127.0.0.1:" + std::to_string(static_cast(wsServer.port()))}; + REQUIRE(backend.waitForConnected()); + + morph::exec::detail::ModelId bound{}; + std::atomic done{false}; + auto observe = [&](morph::exec::detail::ModelId mid) { + bound = mid; + done.store(true); + }; + + SECTION("empty primary, no current instance -> a private instance") { + backend.bindModel(privateBind("SbEchoModel"), callerExec).thenDetached(observe); + REQUIRE(drainUntil(callerExec, done)); + CHECK(bound.v != 0U); + } + + SECTION("non-empty primary -> the server's register-or-attach directory") { + auto first = morph::exec::detail::ModelId{}; + std::atomic firstDone{false}; + backend + .bindModel(morph::backend::detail::BindRequest{.typeId = "SbCounterModel", + .factory = nullptr, + .contextKey = {}, + .primary = "shared-key", + .current = {}}, + callerExec) + .thenDetached([&](morph::exec::detail::ModelId mid) { + first = mid; + firstDone.store(true); + }); + REQUIRE(drainUntil(callerExec, firstDone)); + REQUIRE(first.v != 0U); + + // A second bind on the same key reaches the same live instance. + backend + .bindModel(morph::backend::detail::BindRequest{.typeId = "SbCounterModel", + .factory = nullptr, + .contextKey = {}, + .primary = "shared-key", + .current = {}}, + callerExec) + .thenDetached(observe); + REQUIRE(drainUntil(callerExec, done)); + CHECK(bound == first); + } + + SECTION("non-empty primary plus a current instance -> a re-point") { + auto current = backend.registerModel("SbCounterModel", nullptr); + REQUIRE(current.v != 0U); + backend + .bindModel(morph::backend::detail::BindRequest{.typeId = "SbCounterModel", + .factory = nullptr, + .contextKey = {}, + .primary = "repoint-key", + .current = current}, + callerExec) + .thenDetached(observe); + REQUIRE(drainUntil(callerExec, done)); + CHECK(bound.v != 0U); + } + + SECTION("promoteModel files a live instance and echoes its id back") { + auto mid = backend.registerModel("SbCounterModel", nullptr); + REQUIRE(mid.v != 0U); + backend + .promoteModel( + morph::backend::detail::PromoteRequest{.mid = mid, .typeId = "SbCounterModel", .primary = "promoted"}, + callerExec) + .thenDetached(observe); + REQUIRE(drainUntil(callerExec, done)); + CHECK(bound == mid); + CHECK(backend.listInstances("SbCounterModel") == std::vector{"promoted"}); + } + + SECTION("promoteModel's local guards resolve without touching the wire") { + backend + .promoteModel( + morph::backend::detail::PromoteRequest{ + .mid = morph::exec::detail::ModelId{7}, .typeId = "SbCounterModel", .primary = {}}, + callerExec) + .thenDetached(observe); + REQUIRE(drainUntil(callerExec, done)); + CHECK(bound == morph::exec::detail::ModelId{7}); + } +} + +TEST_CASE("SocketBackend: a bindModel continuation does not run until the caller's executor is pumped", + "[net][socket_backend][registration-surface]") { + // The whole point of the surface: the delivery thread is the caller's + // argument, not the backend's choice. The backend settles from its I/O + // thread; nothing may run on the caller's side until the caller pumps. + morph::exec::ThreadPoolExecutor serverPool{2}; + auto server = std::make_shared(serverPool); + morph::net::SocketServer wsServer{*server, 0}; + REQUIRE(wsServer.listen()); + + // Declared before `backend` -- see the identical comment on the first + // TEST_CASE in this file that needed it (morph#586, data race). + morph::exec::MainThreadExecutor callerExec; + morph::net::SocketBackend backend{"ws://127.0.0.1:" + std::to_string(static_cast(wsServer.port()))}; + REQUIRE(backend.waitForConnected()); + + std::atomic ran{false}; + std::thread::id ranOn{}; + auto completion = backend.bindModel(privateBind("SbEchoModel"), callerExec); + completion.thenDetached([&](morph::exec::detail::ModelId) { + ranOn = std::this_thread::get_id(); + ran.store(true); + }); + + // Give the round trip more than enough time to complete on the io thread. + std::this_thread::sleep_for(std::chrono::milliseconds{200}); + CHECK_FALSE(ran.load()); + + REQUIRE(drainUntil(callerExec, ran)); + CHECK(ranOn == std::this_thread::get_id()); +} + +// The Catch2 assertion macros, not branching logic, push this over the +// cognitive-complexity threshold -- as in the sibling fault-injection cases. +// NOLINTNEXTLINE(readability-function-cognitive-complexity) +TEST_CASE("SocketBackend: a bind settles while the synchronous control channel is still parked", + "[net][socket_backend][registration-surface]") { + // This is the evidence behind morph#569's choice of a native override over + // `SynchronousBackendAdapter`, and behind the deadlock claim in + // docs/spec/core/backend.md. + // + // The documented reconnect hazard is that a control call parks on `_syncCv` + // waiting for a reply only the I/O thread's read loop can deliver -- so + // running one on that thread wedges the transport. A control call that + // never parks cannot have that hazard, whichever thread issues it. Two + // observable consequences prove it does not park: + // + // 1. It is accepted while `sendSync`'s single-call token is held by + // someone else. A blocking bind (the default `bindModel`, or one + // routed through a wrapper's strand) would instead come back with + // `"a synchronous call is already in flight (reentrant use)"`. + // 2. Its reply is delivered, and its `Completion` settles, while that + // other call is *still* parked -- so the bind's settlement does not + // depend on the synchronous channel draining first. + // + // Mutation check: removing `SocketBackend::bindModel` (falling back to the + // default blocking implementation) makes this fail on assertion 1. + FakeWsServer fake; + morph::net::SocketBackend::Config cfg; + cfg.reconnectEnabled = false; + + // Declared before `backend` -- see the identical comment on the first + // TEST_CASE in this file that needed it (morph#586, data race). + morph::exec::MainThreadExecutor callerExec; + morph::net::SocketBackend backend{"ws://127.0.0.1:" + std::to_string(static_cast(fake.port())), cfg}; + fake.acceptAndHandshake(); + REQUIRE(backend.waitForConnected()); + + // Park a legacy synchronous control call: it holds the sendSync token and + // waits on `_syncCv` for a `callId == 0` reply that is deliberately not + // sent until the very end of this test. + std::atomic syncReturned{false}; + std::string syncOutcome; + std::thread syncThread{[&] { + try { + (void)backend.registerModel("SbEchoModel", nullptr); + syncOutcome = "returned"; + } catch (const std::exception& exc) { + // Recorded rather than swallowed: how the parked call ended is not + // what this test asserts, but it is what explains a failure below. + syncOutcome = exc.what(); + } + syncReturned.store(true); + }}; + auto syncEnv = fake.receiveEnvelope(); + REQUIRE(syncEnv.kind == "register"); + REQUIRE(syncEnv.callId == 0U); // the synchronous channel's sentinel + + // (1) The bind is accepted with the token held. + morph::exec::detail::ModelId bound{}; + std::string error; + std::atomic done{false}; + auto completion = backend.bindModel(privateBind("SbEchoModel"), callerExec); + completion + .thenDetached([&](morph::exec::detail::ModelId mid) { + bound = mid; + done.store(true); + }) + .onErrorDetached([&](const std::exception_ptr& exc) { + try { + std::rethrow_exception(exc); + } catch (const std::exception& err) { + error = err.what(); + } + done.store(true); + }); + + // Nothing can settle this bind yet: its reply is only sent below. A + // *blocking* bind, by contrast, has already failed by this point with + // sendSync's reentrant-use error, so pumping the caller's executor here is + // what turns the mutation into an immediate, readable failure rather than + // a hang on the `receiveEnvelope` that follows. + for (int i = 0; i < 20; ++i) { + (void)callerExec.runOnce(); + std::this_thread::sleep_for(std::chrono::milliseconds{5}); + } + bool const accepted = !done.load(); + if (!accepted) { + // Release the parked call and join before failing, so the diagnosis is + // the message below rather than a terminate() on an unjoined thread. + fake.sendFrame(morph::net::detail::WsOpcode::kText, morph::wire::encode(morph::wire::makeOk(0, {}, 7))); + syncThread.join(); + } + INFO("the bind was rejected instead of accepted: " << error); + REQUIRE(accepted); + + auto bindEnv = fake.receiveEnvelope(); + REQUIRE(bindEnv.kind == "register"); + REQUIRE(bindEnv.callId != 0U); // multiplexed, not the synchronous sentinel + + // (2) Answering only the bind settles it, with the sync call still parked. + fake.sendFrame(morph::net::detail::WsOpcode::kText, + morph::wire::encode(morph::wire::makeOk(bindEnv.callId, {}, 4242))); + REQUIRE(drainUntil(callerExec, done)); + CHECK(error.empty()); + CHECK(bound == morph::exec::detail::ModelId{4242}); + CHECK_FALSE(syncReturned.load()); + + // Release the parked call so the thread can be joined. + fake.sendFrame(morph::net::detail::WsOpcode::kText, morph::wire::encode(morph::wire::makeOk(0, {}, 7))); + syncThread.join(); + CHECK(syncOutcome == "returned"); +} + +// NOLINTNEXTLINE(readability-function-cognitive-complexity) +TEST_CASE("SocketBackend: several binds are in flight at once and are matched by callId with replies out of order", + "[net][socket_backend][registration-surface]") { + // The corollary of "a bind never parks": several are outstanding at once. + // Under the default blocking `bindModel` the first call here would never + // return (its reply is only sent after all four have been issued), so this + // test fails by hanging if the native override is removed. + FakeWsServer fake; + morph::net::SocketBackend::Config cfg; + cfg.reconnectEnabled = false; + + // Declared before `backend` -- see the identical comment on the first + // TEST_CASE in this file that needed it (morph#586, data race). + morph::exec::MainThreadExecutor callerExec; + morph::net::SocketBackend backend{"ws://127.0.0.1:" + std::to_string(static_cast(fake.port())), cfg}; + fake.acceptAndHandshake(); + REQUIRE(backend.waitForConnected()); + + constexpr int kBinds = 4; + std::vector> completions; + std::vector bound(kBinds); + std::atomic settled{0}; + for (int i = 0; i < kBinds; ++i) { + completions.push_back(backend.bindModel(privateBind("SbEchoModel"), callerExec)); + completions.back().thenDetached([&, i](morph::exec::detail::ModelId mid) { + bound[static_cast(i)] = mid; + settled.fetch_add(1); + }); + } + + std::vector callIds; + for (int i = 0; i < kBinds; ++i) { + auto env = fake.receiveEnvelope(); + REQUIRE(env.callId != 0U); + callIds.push_back(env.callId); + } + REQUIRE(std::set(callIds.begin(), callIds.end()).size() == kBinds); + + // Answered back to front: the reply router, not arrival order, decides + // which Completion each id settles. + for (int i = kBinds - 1; i >= 0; --i) { + fake.sendFrame(morph::net::detail::WsOpcode::kText, + morph::wire::encode(morph::wire::makeOk(callIds[static_cast(i)], {}, + static_cast(100 + i)))); + } + + std::atomic allDone{false}; + for (int i = 0; i < 500 && !allDone.load(); ++i) { + if (!callerExec.runOnce()) { + std::this_thread::sleep_for(std::chrono::milliseconds{5}); + } + allDone.store(settled.load() == kBinds); + } + REQUIRE(allDone.load()); + for (int i = 0; i < kBinds; ++i) { + CHECK(bound[static_cast(i)] == morph::exec::detail::ModelId{static_cast(100 + i)}); + } +} + +TEST_CASE("SocketBackend: a bind on a dropped connection is rejected rather than left unsettled", + "[net][socket_backend][registration-surface]") { + morph::exec::MainThreadExecutor callerExec; + std::string error; + std::atomic done{false}; + auto capture = [&](const std::exception_ptr& exc) { + try { + std::rethrow_exception(exc); + } catch (const std::exception& err) { + error = err.what(); + } + done.store(true); + }; + + SECTION("issued while already disconnected") { + morph::net::SocketBackend::Config cfg; + cfg.reconnectEnabled = false; + cfg.connectTimeout = std::chrono::milliseconds{200}; + morph::net::SocketBackend backend{"ws://127.0.0.1:1", cfg}; // nothing listens there + REQUIRE_FALSE(backend.waitForConnected(std::chrono::milliseconds{300})); + backend.bindModel(privateBind("SbEchoModel"), callerExec).onErrorDetached(capture); + REQUIRE(drainUntil(callerExec, done)); + CHECK_THAT(error, Catch::Matchers::ContainsSubstring("disconnected")); + } + + SECTION("dropped while in flight") { + auto fake = std::make_unique(); + morph::net::SocketBackend::Config cfg; + cfg.reconnectEnabled = false; + morph::net::SocketBackend backend{"ws://127.0.0.1:" + std::to_string(static_cast(fake->port())), + cfg}; + fake->acceptAndHandshake(); + REQUIRE(backend.waitForConnected()); + + backend.bindModel(privateBind("SbEchoModel"), callerExec).onErrorDetached(capture); + REQUIRE(fake->receiveEnvelope().callId != 0U); + fake.reset(); // the peer goes away without ever replying + REQUIRE(drainUntil(callerExec, done)); + CHECK_THAT(error, Catch::Matchers::ContainsSubstring("disconnected")); + } +} + +TEST_CASE("SocketBackend: a bind rejected by the server surfaces the server's own message", + "[net][socket_backend][registration-surface]") { + morph::exec::ThreadPoolExecutor serverPool{2}; + auto authz = std::make_shared(); + auto server = std::make_shared(serverPool, authz); + morph::net::SocketServer wsServer{*server, 0}; + REQUIRE(wsServer.listen()); + + // Declared before `backend` -- see the identical comment on the first + // TEST_CASE in this file that needed it (morph#586, data race). + morph::exec::MainThreadExecutor callerExec; + morph::net::SocketBackend backend{"ws://127.0.0.1:" + std::to_string(static_cast(wsServer.port()))}; + REQUIRE(backend.waitForConnected()); + + std::string error; + std::atomic done{false}; + backend.bindModel(privateBind("SbEchoModel"), callerExec).onErrorDetached([&](const std::exception_ptr& exc) { + try { + std::rethrow_exception(exc); + } catch (const std::exception& err) { + error = err.what(); + } + done.store(true); + }); + REQUIRE(drainUntil(callerExec, done)); + // Same " failed: " wording the blocking verbs use. + CHECK_THAT(error, Catch::Matchers::ContainsSubstring("register failed: unauthorized")); +} + +TEST_CASE("SocketBackend: a reconnect handler can re-bind through the structural surface without waiting", + "[net][socket_backend][disconnect][registration-surface]") { + // The shape a reconnect handler takes once its caller is on the structural + // surface (morph#570): issue the bind, attach a continuation, return. The + // handler parks on nothing, so it has no reply to wait for and cannot hold + // up whichever thread runs it. + // + // Note what this test does *not* claim: it does not show the dedicated + // handler thread has become unnecessary. `Bridge`'s reconnect handler still + // calls the blocking verbs, and that is what keeps the thread load-bearing + // -- see docs/spec/core/backend.md. + ReconnectFixture fixture; + morph::exec::ThreadPoolExecutor callerPool{1}; + std::atomic handlerRan{false}; + std::atomic reboundOk{false}; + + fixture.bounce([&] { + handlerRan.store(true); + fixture.backend->bindModel(privateBind("SbEchoModel"), callerPool) + .thenDetached([&](morph::exec::detail::ModelId mid) { reboundOk.store(mid.v != 0U); }); + }); + + spinUntil([&] { return reboundOk.load(); }, 500); + CHECK(handlerRan.load()); + CHECK(reboundOk.load()); + + // The transport is unharmed: the synchronous channel was never involved. + CHECK(fixture.backend->registerModel("SbEchoModel", nullptr).v != 0U); +}