From 45021688a1bfb45d17333f97c5f774782de72b9d Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Sun, 4 Oct 2026 12:29:56 +0200 Subject: [PATCH 1/2] core, wire: a hello-negotiated cancel envelope that RemoteServer maps onto the run's stop source (#864) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A `cancel` names the execute it stops by `cancelCallId` and carries its own `callId`, so the server's `ok` to it can never be matched against the still pending execute. The server advertises the kind in its `hello` reply (`ProtocolRange::capabilities`, "cancel"); a client sends it only after such a reply, so a server that predates it never sees one. Both members are additive, so this is a capability, not a kProtocolVersion bump. RemoteServer files each admitted Task execute that arrived on a connection scope, with what its admission decided on: the verified principal, the model and action types, the instance and its owner. A cancel is stamped (stampVerifiedPrincipal), looked up only under the connection it arrived on, and honoured only when its verified principal is the execute's and it passes the execute's own authorize and authorizeInstance. Every outcome — stopped, unknown, finished, another connection's, another principal's, refused, unscoped — answers the same `ok`, so a cancel cannot probe for calls. Co-Authored-By: Claude Opus 5.5 --- docs/spec/core/backend.md | 12 +- docs/spec/core/wire.md | 123 ++++++++- docs/spec/security.md | 9 + include/morph/core/remote.hpp | 169 +++++++++++-- include/morph/core/wire.hpp | 80 +++++- tests/CMakeLists.txt | 1 + tests/test_remote_cancel.cpp | 386 +++++++++++++++++++++++++++++ tests/test_wire_omitted_fields.cpp | 2 + 8 files changed, 751 insertions(+), 31 deletions(-) create mode 100644 tests/test_remote_cancel.cpp diff --git a/docs/spec/core/backend.md b/docs/spec/core/backend.md index cd0eadd0f..bc1508ab1 100644 --- a/docs/spec/core/backend.md +++ b/docs/spec/core/backend.md @@ -589,7 +589,8 @@ external reference is dropped. | `register` | `typeId`, `[contextKey]` | `ok` with `modelId` (body empty) | Authenticates the caller (`_authorizer->authenticate(env.session)`), stamping the verified principal onto `env.session.principal` (clearing it when unauthenticated) exactly as `execute` does, then consults `_authorizer->authorizeRegister(env.session, typeId)` — a `false` reply is `err "unauthorized"` and **no instance is created**. Only then creates the model via the `ModelRegistryFactory` and records the (already-verified) principal as its owner. Empty `typeId` → `err "register requires a typeId"` (checked before authorization). If `contextKey` is non-empty, consults `ServerConfig::logProvider` (if set) and, when it returns a non-null log, calls `holder->attachActionLog(log, contextKey)`. The assigned `modelId` is an **opaque** (non-sequential) value — see below. | | `deregister` | `modelId` | `ok` or `err` | Consults `authorizeInstance` against the recorded owner (denied → `err "unauthorized"`); otherwise erases the model and its owner entry from the registry. | | `execute` | `modelId`, `modelType`, `actionType`, `body`, `session` | `ok` with `body` or `err` | See the execute flow below. | -| `hello` | `protocolVersion` | `ok` with `body` = `ProtocolRange`, or `err "protocol version unsupported"` | Protocol-version negotiation, exchanged once per connection before any `register`/`execute`. Carries no `session` and is not authorized — orthogonal to `IAuthorizer`. See [wire.md](wire.md#protocol-version-negotiation). | +| `hello` | `protocolVersion` | `ok` with `body` = `ProtocolRange`, or `err "protocol version unsupported"` | Protocol-version negotiation, exchanged once per connection before any `register`/`execute`. Carries no `session` and is not authorized — orthogonal to `IAuthorizer`. The range's `capabilities` lists `"cancel"`. See [wire.md](wire.md#protocol-version-negotiation). | +| `cancel` | `cancelCallId`, `session` | `ok` (the cancel's own `callId`), always | Stamps the verified principal, then requests stop on the run of the `execute` filed under `cancelCallId` **on the same connection**, if the cancel's verified principal is the execute's and the cancel passes the execute's own `authorize` and `authorizeInstance`. Anything else — unknown, finished, another connection's, another principal's, refused, unscoped — is the same `ok` and changes nothing. See [wire.md](wire.md#cancelling-a-call). | | `schemas` | `typeId`, `session` | `ok` with `body` = `{actionType: schema}`, or `err` | Empty `typeId` → `err "schemas requires a typeId"` (checked before authorization). Authenticates, then consults `_authorizer->authorize(env.session, typeId, {})` — the same type-level read hook `instances` uses; denied → `err "unauthorized"`. Answers from `ActionDispatcher::schemasJson(typeId)`; a type with no registered actions yields `{}`, not an error. See [Serving action schemas](#serving-action-schemas). | **Execute flow.** Admission runs on the server strand (`dispatchExecute`), the @@ -630,7 +631,9 @@ action on the model's strand. Admission, in order: 6. **Payload completeness** (opt-in; see [`PayloadCompleteness`](#payloadcompleteness--enforcing-the-action-evolution-policy)). 7. **Reserve the in-flight slot**, arm `LimitPolicy::executeTimeout` if one is - configured, and **post to the model's strand** a task that enters the + configured, file the call as cancellable if its handler is a Task and it + arrived on a connection scope (see the `cancel` row above), and **post to + the model's strand** a task that enters the instance's action gate, installs a `ScopedContext` from the (now possibly rewritten) `env.session`, calls `dispatch(modelType, actionType, *holder, body)` on the server's dispatcher, and replies `ok` with the serialised @@ -642,7 +645,8 @@ action on the model's strand. Admission, in order: The reply is sent exactly once, by whichever of the model strand's finish and the `executeTimeout` gets there first. Before sending it, that path posts the -in-flight decrement back to the server strand, so a `health()` or +in-flight decrement — and the call's withdrawal from the cancellable calls — +back to the server strand, so a `health()` or `drainedWithin()` asked by someone who has seen the reply already counts it as finished. @@ -682,7 +686,7 @@ connection scope — see "Connection scopes" below. **`handleInline(msg)`** — synchronous entry point for control envelopes (`register`, `deregister`, `attach`, `assign`, `instances`, `schemas`, -`hello`), the path `SimulatedRemoteBackend` uses. It posts the envelope to the +`hello`, `cancel`), the path `SimulatedRemoteBackend` uses. It posts the envelope to the server strand and blocks until the reply is written; when the caller is already on the server strand — host code the strand itself is running, such as a model constructor or a `LogProvider` that registers another model — it runs diff --git a/docs/spec/core/wire.md b/docs/spec/core/wire.md index 0027ab667..2e5b683ca 100644 --- a/docs/spec/core/wire.md +++ b/docs/spec/core/wire.md @@ -12,6 +12,7 @@ carries all request and reply variants, discriminated by a `kind` string field. - [Omitted default fields](#omitted-default-fields) - [Parsing guarantees and hardening](#parsing-guarantees-and-hardening) - [Protocol version negotiation](#protocol-version-negotiation) +- [Cancelling a call](#cancelling-a-call) - [Serving action schemas](#serving-action-schemas) - [Action-evolution policy](#action-evolution-policy) - [API reference](#api-reference) @@ -35,6 +36,7 @@ their `kind` needs and leave the rest as default-constructed values. | `"deregister"` | request | Client destroys an instance. | `modelId` | | `"execute"` | request | Client dispatches an action. | `callId`, `modelId`, `modelType`, `actionType`, `body`, `session` | | `"hello"` | request | Client announces its protocol version, once per connection, before any `register`/`execute`. See [Protocol version negotiation](#protocol-version-negotiation). | `protocolVersion` | +| `"cancel"` | request | Client asks the server to stop one of its own in-flight `execute` calls. Replies `ok` whatever became of the call. Sent only to a server whose `hello` reply advertised `"cancel"`. See [Cancelling a call](#cancelling-a-call). | `callId` (the cancel's own), `cancelCallId` (the execute's), `session` | | `"ok"` | reply | Server success. | `callId`, `body` (serialized result, or — for a `"hello"` reply — the server's `ProtocolRange`), `modelId` (for register-replies) | | `"err"` | reply | Server failure. | `callId`, `message` | @@ -62,7 +64,7 @@ session — they round-trip it verbatim; enforcement lives in the server. ## Factory functions -Ten free functions construct `Envelope` instances with the correct `kind` and +Eleven free functions construct `Envelope` instances with the correct `kind` and relevant fields. Callers never set `kind` manually. | Function | `kind` | Parameters | @@ -75,6 +77,7 @@ relevant fields. Callers never set `kind` manually. | `makeSchemas(typeId)` | `"schemas"` | Model type id whose action descriptions are wanted. See [Serving action schemas](#serving-action-schemas). | | `makeDeregister(modelId)` | `"deregister"` | Instance id to destroy. | | `makeHello(protocolVersion = kProtocolVersion)` | `"hello"` | Protocol version the sender speaks. See [Protocol version negotiation](#protocol-version-negotiation). | +| `makeCancel(targetCallId)` | `"cancel"` | `callId` of the `execute` to stop, stored in `cancelCallId`. The caller assigns the cancel's own `callId` and `session`. See [Cancelling a call](#cancelling-a-call). | | `makeOk(callId = 0, body = {}, modelId = 0)` | `"ok"` | Correlation id, serialized result (stored in the `body` field), optional model id (for register-replies). | | `makeErr(message, callId = 0)` | `"err"` | Error message, optional correlation id. | @@ -90,8 +93,9 @@ it internally. | `decode` | `Envelope decode(std::string_view)` | Deserializes from JSON via `glz::read<{.error_on_unknown_keys = false}>`. Rejects input longer than `kMaxEnvelopeBytes` and throws `std::runtime_error` on an oversized or syntactically malformed envelope. **Ignores unknown/extra keys** (forward compatibility) and **does not reject duplicate JSON keys** — see [Parsing guarantees and hardening](#parsing-guarantees-and-hardening). | glaze reflects the struct's public members, so the JSON object keys are exactly -the C++ field names (`kind`, `callId`, `typeId`, `contextKey`, `modelId`, -`modelType`, `actionType`, `body`, `message`, `session`). `decode` starts from a +the C++ field names (`kind`, `callId`, `typeId`, `contextKey`, `primary`, +`shared`, `modelId`, `modelType`, `actionType`, `body`, `message`, `session`, +`protocolVersion`, `cancelCallId`). `decode` starts from a default-constructed `Envelope`, so any key absent from the input JSON keeps its default value — omitting fields a given `kind` does not use is expected and does not throw. `decode` reads with `error_on_unknown_keys = false`, so an @@ -104,7 +108,7 @@ turn it into an `"err"` reply rather than propagating it (see ### Omitted default fields -`Envelope` is a union of all kinds: an `ok` reply uses three of its thirteen +`Envelope` is a union of all kinds: an `ok` reply uses three of its fourteen members and a `deregister` request uses two. glaze writes every member of a struct it is handed, so without the omission rule below a minimal `ok` reply carrying an 8-byte payload is **255 bytes, 213 of them fields the kind does not @@ -213,7 +217,7 @@ notice", and a second implementation is the thing that finds the difference. The failure mode the design has to survive is a member being added or renamed without this list following it. Both are compile errors: -`detail::defaultValuedKeys` `static_assert`s that `Envelope` still has thirteen +`detail::defaultValuedKeys` `static_assert`s that `Envelope` still has fourteen reflected members, and that every name in `detail::kOmittableEnvelopeKeys` is one of the reflected keys — a rename would otherwise make glaze report `unknown_key` and turn *every* `encode` call into a throw at runtime. @@ -434,6 +438,107 @@ no version check, `protocolVersion` stays `0` on every envelope. clear `"protocol version unsupported"` refusal at connect time instead of a confusing failure on the first `execute`. +## Cancelling a call + +A client that stops waiting for an `execute` — its deadline passed, its caller +was stopped, its bridge went away — tells the server with a `"cancel"`, so a +handler that can stop does, rather than running on for a reply nobody reads. + +| `kind` | Direction | Fields used | Reply | +|---|---|---|---| +| `"cancel"` | request | `callId` (the cancel's own), `cancelCallId` (the `execute`'s `callId`), `session` | `"ok"` with the cancel's own `callId`, always | + +### Negotiated through `"hello"`, not through a version bump + +A server that honours `"cancel"` lists it in the `capabilities` array of the +`ProtocolRange` its `"hello"` reply carries: + +``` +{"min":1,"max":1,"capabilities":["cancel"]} +``` + +A client sends `"cancel"` only after a `"hello"` reply whose list names it +(`wire::helloAdvertises(reply, wire::kCapabilityCancel)`). Every other answer +— a legacy peer's `err "unknown envelope kind: hello"`, an `ok` whose body has +no `capabilities` key, a client that never negotiated — means it never sends +one. So a server that predates `"cancel"` never receives one, and keeps working +exactly as before with a client that knows the kind. + +This is a capability, not a `kProtocolVersion` bump, for the reason the +[Action-evolution policy](#action-evolution-policy) sets the bar where it does: +nothing is removed or retyped. The new `Envelope` member (`cancelCallId`) and +the new `ProtocolRange` member (`capabilities`) are both additive, and an +older decoder that ignores unknown keys reads either message unchanged. A +version bump would also have refused every client still announcing version +`1` under the default `{kProtocolVersion, kProtocolVersion}` range, which a +feature an old client simply does not use gives no reason to do. + +### Why the target travels in `cancelCallId` + +A `"cancel"` is a request with a reply of its own, and every reply is routed by +its `callId` (see [Omitted default fields](#omitted-default-fields)). Had the +target's id travelled in `callId`, the server's `ok` to the cancel would be +matched by the client against the execute still pending under that id and +settle it with an empty result. With its own `callId` the cancel's reply is +routed like any other: a client that files nothing under that id drops it. + +`cancelCallId` is payload, not identity, so `encode` omits it at its default +`0`, like every other addressing field the kind does not use. + +### What the server does with it + +`RemoteServer` keeps, for each admitted `execute` whose handler returns +`core::async::Task` and which arrived on a connection scope +([backend.md](backend.md#connection-scopes)), the run's stop source together +with what the execute was authorized on: its verified principal, its model and +action types, its instance and that instance's recorded owner. On a +`"cancel"`, on the server strand: + +1. It stamps the verified principal onto the cancel's session + (`stampVerifiedPrincipal`), exactly as for `execute` and `deregister`. +2. It looks the call up under **the connection the cancel arrived on** and + `cancelCallId`. A connection reaches only its own calls: `callId`s are + chosen by each client and are not unique across clients. +3. It requires the cancel's verified principal to equal the execute's, then + runs the execute's own gates as the cancel's caller — + `authorize(session, modelType, actionType)` and + `authorizeInstance(session, modelType, actionType, modelId, owner)`. +4. Only if all of that holds does it request stop on the run's stop source + and forget the call. + +Then it replies `ok` with the cancel's `callId` — **the same reply in every +case**: the call was stopped, the id was never used, the call already +finished, the call belongs to another connection or another principal, or a +gate refused. A cancel therefore cannot be used to learn which calls exist or +whose they are. + +The stop is the same one `LimitPolicy::executeTimeout` requests: a Task +handler unwinds at its next stop-aware `co_await` with +`core::async::OperationCancelled`, and the call replies `err` with that +exception's message, which a client that already gave up on the call drops. A +handler that is not a Task cannot be stopped, so its call is never filed; a +cancel naming it is a no-op like any other. A call still queued behind another +on its instance's action gate is stopped before it starts: its handler's token +is already stopped when it runs. + +An unscoped message (`ConnectionId` `0`, the two-argument `handle()`) names no +call: the server files no unscoped call as cancellable, because unscoped +callers share one id space and could not be told apart. + +### Why the principal check, on top of the connection + +The connection alone would let one principal stop another's call on a +connection several principals share — a gateway multiplexing its users onto +one socket — wherever `authorizeInstance` admits both, as it does for an +unowned or shared instance. Stamping alone would not close that either: the +comparison has to be with the principal *the execute* was admitted under, +which is why it is recorded at admission rather than recomputed. This is the +shape of the two `deregister` holes the server already closes: a request +checked against the client's claimed identity, and a connection reaching a +hold that is not its own. `tests/test_remote_cancel.cpp` pins both directions: +a cancel from another connection, and a cancel carrying a forged claim on the +owning connection, each leave the call to complete with its result. + ## Serving action schemas `morph::forms::schemaJson()` renders one action as a JSON Schema document — @@ -680,6 +785,7 @@ client. | `message` | `std::string` | `""` | `"err"` — free-text error message. | | `session` | `::morph::session::Context` | default | `"execute"` — authorization and routing context. | | `protocolVersion` | `uint32_t` | `0` | `"hello"` — protocol version the sender speaks. `0` means unspecified/legacy peer; not otherwise inspected. | +| `cancelCallId` | `uint64_t` | `0` | `"cancel"` — the `callId` of the `execute` to stop. `0` names no call. | ### Factory functions @@ -693,6 +799,7 @@ client. | `makeSchemas` | `Envelope makeSchemas(std::string typeId)` | | `makeDeregister` | `Envelope makeDeregister(uint64_t modelId)` | | `makeHello` | `Envelope makeHello(uint32_t protocolVersion = kProtocolVersion)` | +| `makeCancel` | `Envelope makeCancel(uint64_t targetCallId)` | | `makeOk` | `Envelope makeOk(uint64_t callId = 0, std::string body = {}, uint64_t modelId = 0)` | | `makeErr` | `Envelope makeErr(std::string message, uint64_t callId = 0)` | @@ -700,16 +807,18 @@ client. | Symbol | Signature / shape | Notes | |---|---|---| -| `ProtocolRange` | `struct { uint32_t min = kProtocolVersion; uint32_t max = kProtocolVersion; }` | A server's supported version range; serialized into a `"hello"` `"ok"` reply's `body`. | +| `ProtocolRange` | `struct { uint32_t min = kProtocolVersion; uint32_t max = kProtocolVersion; std::vector capabilities; }` | A server's supported version range and the optional kinds it honours; serialized into a `"hello"` `"ok"` reply's `body`. | | `ProtocolNegotiationResult` | `enum class : uint8_t { Negotiated, LegacyPeer }` | Outcome of `interpretHelloReply`. | | `interpretHelloReply` | `ProtocolNegotiationResult interpretHelloReply(const Envelope& reply)` | Throws `std::runtime_error` if `reply` is an `"err"` other than `"unknown envelope kind: hello"`. | +| `helloAdvertises` | `bool helloAdvertises(const Envelope& reply, std::string_view capability)` | `true` only for an `"ok"` whose `ProtocolRange` lists `capability`; never throws on a body it cannot read. | +| `kCapabilityCancel` | `std::string_view` = `"cancel"` | The capability a server lists when it honours `"cancel"`. | ### Serialization | Symbol | Signature | Throws | |---|---|---| | `encode` | `std::string encode(const Envelope&)` | `std::runtime_error` on serialisation failure | -| `detail::kOmittableEnvelopeKeys` | `constexpr std::array` | — the envelope keys `encode` leaves out when the member is at its default (everything but `kind`). | +| `detail::kOmittableEnvelopeKeys` | `constexpr std::array` | — the envelope keys `encode` leaves out when the member is at its default (everything but `kind` and `callId`). | | `detail::defaultValuedKeys` | `std::span defaultValuedKeys(const Envelope&, std::array&)` | — which of those keys this envelope omits; writes into the caller's array so the per-message path allocates nothing. | | `detail::isDefaultSession` | `bool isDefaultSession(const ::morph::session::Context&)` | — whether the whole `session` object can be left out. | | `decode` | `Envelope decode(std::string_view)` | `std::runtime_error` if the input exceeds `kMaxEnvelopeBytes` or is a syntactically malformed envelope. Unknown/extra keys are **ignored** (`error_on_unknown_keys = false`); duplicate keys do **not** throw (last-wins) — see [Parsing guarantees and hardening](#parsing-guarantees-and-hardening). | diff --git a/docs/spec/security.md b/docs/spec/security.md index ddc51efdd..aec016a26 100644 --- a/docs/spec/security.md +++ b/docs/spec/security.md @@ -412,6 +412,15 @@ method: and the instance is left intact. The stamp is what makes the ownership comparison sound: without it, a caller could write the owner's name into `principal` and release the owner's instance with any token, or none. +- **On `cancel`** — `RemoteServer` stamps the verified principal onto the + cancel's session and looks the target `execute` up only among the calls + **of the connection the cancel arrived on**. It then requires the cancel's + verified principal to equal the one the `execute` was admitted under, and + runs the execute's own `authorize(session, modelType, actionType)` and + `authorizeInstance(session, modelType, actionType, modelId, ownerPrincipal)` + as the cancel's caller. Only then is the call stopped. Every refusal answers + the same `ok` an unknown or finished call gets, so a cancel cannot probe for + other callers' calls. See [wire.md](core/wire.md#cancelling-a-call). ### Backward compatibility diff --git a/include/morph/core/remote.hpp b/include/morph/core/remote.hpp index f796f256c..85082ebc2 100644 --- a/include/morph/core/remote.hpp +++ b/include/morph/core/remote.hpp @@ -759,6 +759,88 @@ class RemoteServer : public std::enable_shared_from_this { } } + /// One admitted execute, as `callFinished` and a cancel's lookup know it. + struct CallTag { + /// Connection the execute arrived on; `0` for unscoped. + ConnectionId cid{0}; + /// The execute's own `callId`, as the client chose it. + std::uint64_t callId{0}; + /// The run's stop source when the call is filed as cancellable, else + /// null. Held, not just compared: it is what tells this call's entry + /// from a later call's under a reused `callId`. + std::shared_ptr<::core::async::StopSource> stop; + }; + + /// Ends one admitted execute's bookkeeping once its reply has been sent: + /// withdraws it from the cancellable calls, then counts it finished. On + /// the server strand. + /// @param tag The finished call. + void callFinished(const CallTag& tag) { + if (tag.stop) { + auto const found = _cancellable.find(CancelKey{.cid = tag.cid, .callId = tag.callId}); + if (found != _cancellable.end() && found->second.stop == tag.stop) { + _cancellable.erase(found); + } + } + executeFinished(); + } + + /// Files an admitted Task execute so the caller that made it can cancel + /// it. On the server strand, after the execute's own authorization. + /// + /// Keeps what that authorization decided on — the verified principal, the + /// model and action types, the instance and its recorded owner — so a + /// cancel is judged by the same gates, against the same facts, as the + /// execute it stops. A second execute under the same `callId` on the same + /// connection replaces the first's entry; `callFinished` removes only its + /// own. + /// @param tag The call, with its stop source. + /// @param env The admitted execute, its principal already stamped. + /// @param owner The instance's recorded owner. + void fileCancellable(const CallTag& tag, const ::morph::wire::Envelope& env, const std::string& owner) { + _cancellable.insert_or_assign(CancelKey{.cid = tag.cid, .callId = tag.callId}, + CancelTarget{.stop = tag.stop, + .principal = env.session.principal, + .modelType = env.modelType, + .actionType = env.actionType, + .modelId = env.modelId, + .owner = owner}); + } + + /// Requests stop on the execute a `cancel` names, if the cancel may stop + /// it; otherwise does nothing. On the server strand. + /// + /// The call is looked up under the connection the cancel arrived on, so a + /// connection reaches only its own calls. The cancel must then carry the + /// execute's verified principal, and pass the execute's own gates — + /// `authorize` for its types and `authorizeInstance` for its instance — + /// as the cancel's caller. Every refusal is silent: the reply is the same + /// `ok` an unknown or finished call gets. + /// @param env The cancel, its principal already stamped. + /// @param cid Connection it arrived on; `0` reaches no call. + void cancelCall(const ::morph::wire::Envelope& env, ConnectionId cid) { + if (cid == 0) { + return; + } + auto const found = _cancellable.find(CancelKey{.cid = cid, .callId = env.cancelCallId}); + if (found == _cancellable.end()) { + return; + } + CancelTarget const& target = found->second; + if (env.session.principal != target.principal || + !_authorizer->authorize(env.session, target.modelType, target.actionType) || + !_authorizer->authorizeInstance(env.session, target.modelType, target.actionType, target.modelId, + target.owner)) { + return; + } + auto const stop = target.stop; + _cancellable.erase(found); + // Runs the handler's stop callbacks here, on the server strand; each + // resumes its coroutine through the handler's own resumer, on the + // model's strand, as the execute timeout's stop does from its thread. + static_cast(stop->request_stop()); + } + /// @brief Authenticates @p env's session and makes the verified identity /// authoritative on it. /// @@ -1267,7 +1349,17 @@ class RemoteServer : public std::enable_shared_from_this { releaseScoped(mid, cid); reply(::morph::wire::encode(::morph::wire::makeOk(env.callId))); } else if (env.kind == "execute") { - dispatchExecute(std::move(env), reply); + dispatchExecute(std::move(env), reply, cid); + } else if (env.kind == "cancel") { + // Stamped before the ownership comparison in cancelCall, for + // the reason deregister is: the comparison keys on the + // verified principal, never on the client's claim. + stampVerifiedPrincipal(env); + cancelCall(env, cid); + // One answer whatever became of the call -- stopped, unknown, + // finished, or another caller's -- so a cancel cannot be used + // to learn which calls exist. + reply(::morph::wire::encode(::morph::wire::makeOk(env.callId))); } else if (env.kind == "hello") { const std::uint32_t minV = _config.minProtocolVersion; const std::uint32_t maxV = _config.maxProtocolVersion; @@ -1275,7 +1367,10 @@ class RemoteServer : public std::enable_shared_from_this { reply(::morph::wire::encode(::morph::wire::makeErr("protocol version unsupported", env.callId))); } else { std::string body; - (void)glz::write_json(::morph::wire::ProtocolRange{.min = minV, .max = maxV}, body); + (void)glz::write_json( + ::morph::wire::ProtocolRange{ + .min = minV, .max = maxV, .capabilities = {std::string{::morph::wire::kCapabilityCancel}}}, + body); reply(::morph::wire::encode(::morph::wire::makeOk(env.callId, std::move(body)))); } } else { @@ -1298,7 +1393,7 @@ class RemoteServer : public std::enable_shared_from_this { // docs/spec/security.md), so it is deliberately not broken up. On the // server strand; the admitted run is posted to the model's strand. // NOLINTNEXTLINE(readability-function-cognitive-complexity) - void dispatchExecute(::morph::wire::Envelope env, std::function reply) { + void dispatchExecute(::morph::wire::Envelope env, std::function reply, ConnectionId cid) { // The admission phase only, on the server strand: the run itself is // posted to the model's strand and zoned there, in startRemote. MORPH_ZONE("RemoteServer::dispatchExecute"); @@ -1371,6 +1466,21 @@ class RemoteServer : public std::enable_shared_from_this { _inFlight += 1; ::morph::observe::detail::emitMetric(::morph::observe::Metric::executeInFlight, static_cast(_inFlight)); + // The stop source a Task handler's token comes from. Requested by the + // timeout alongside its reply, so a Task handler still suspended when + // the caller is answered unwinds and leaves the action gate rather than + // holding the model until its await completes; and by a `cancel` from + // the caller that made the call. A call is cancellable only when its + // handler is a Task, the one kind a stop reaches, and only on a + // connection scope: an unscoped message has no connection to name the + // call by, and callIds are the client's own, not unique across clients. + bool const taskHandler = _dispatcher.dispatchesAsync(env.modelType, env.actionType); + bool const cancellable = taskHandler && cid != 0; + std::shared_ptr<::core::async::StopSource> stopSource; + if (_executeTimeouts || cancellable) { + stopSource = std::make_shared<::core::async::StopSource>(); + } + CallTag const tag{.cid = cid, .callId = env.callId, .stop = cancellable ? stopSource : nullptr}; auto self = shared_from_this(); auto finished = std::make_shared(); // Gives the slot back if this frame leaves by exception before the run @@ -1380,39 +1490,38 @@ class RemoteServer : public std::enable_shared_from_this { struct Reservation { RemoteServer& server; std::shared_ptr finished; + CallTag tag; bool handedOff = false; - Reservation(RemoteServer& owner, std::shared_ptr flag) - : server{owner}, finished{std::move(flag)} {} + Reservation(RemoteServer& owner, std::shared_ptr flag, CallTag call) + : server{owner}, finished{std::move(flag)}, tag{std::move(call)} {} Reservation(const Reservation&) = delete; Reservation& operator=(const Reservation&) = delete; Reservation(Reservation&&) = delete; Reservation& operator=(Reservation&&) = delete; - // NOLINTNEXTLINE(bugprone-exception-escape): executeFinished runs the server's drain bookkeeping on its owner; a failure there has no recovery in a destructor. + // NOLINTNEXTLINE(bugprone-exception-escape): callFinished runs the server's drain bookkeeping on its owner; a failure there has no recovery in a destructor. ~Reservation() { if (!handedOff && !finished->test_and_set()) { - server.executeFinished(); + server.callFinished(tag); } } - } reservation{*this, finished}; + } reservation{*this, finished, tag}; + if (cancellable) { + fileCancellable(tag, env, inst->owner); + } std::uint64_t const callId = env.callId; auto replySlot = std::make_shared>(std::move(reply)); // The decrement is posted before the reply goes out, so anything posted // to the server strand by someone who has seen the reply runs after it. - auto complete = [self, finished, replySlot](std::string msg) { + auto const complete = [self, finished, replySlot, tag](std::string msg) { if (!finished->test_and_set()) { - self->_strand.postTask([self] { self->executeFinished(); }); + self->_strand.postTask([self, tag] { self->callFinished(tag); }); (*replySlot)(std::move(msg)); } }; ::morph::async::detail::TimeoutScheduler::Handle timeoutHandle{}; - // Requested by the timeout alongside its reply, so a Task handler still - // suspended when the caller is answered unwinds and leaves the action - // gate rather than holding the model until its await completes. - std::shared_ptr<::core::async::StopSource> stopSource; if (_executeTimeouts) { - stopSource = std::make_shared<::core::async::StopSource>(); // Safe to fire after its own `cancel()`, which the model strand's // finish issues on both its success and its failure arm: // `TimeoutScheduler::cancel` stops a callback that has not started @@ -1443,7 +1552,7 @@ class RemoteServer : public std::enable_shared_from_this { // Through the instance's action gate: an action starts only once the one // before it has finished, which a Task handler does when its Task // completes rather than when the strand task that started it returns. - if (_dispatcher.dispatchesAsync(run.env.modelType, run.env.actionType)) { + if (taskHandler) { // A Task run is shared: its completion callback outlives the strand // task. auto shared = std::make_shared(std::move(run)); @@ -1655,6 +1764,34 @@ class RemoteServer : public std::enable_shared_from_this { std::unordered_map> _connectionScopes; + /// Names one cancellable call: the connection it arrived on and the + /// `callId` its client chose. + struct CancelKey { + ConnectionId cid{0}; + std::uint64_t callId{0}; + bool operator==(const CancelKey&) const = default; + }; + struct CancelKeyHash { + std::size_t operator()(const CancelKey& key) const noexcept { + // Golden-ratio multiply spreads the connection id before the mix, + // so two connections' runs of small callIds do not collide. + constexpr std::uint64_t kSpread = 0x9E3779B97F4A7C15ULL; + return std::hash{}((key.cid * kSpread) ^ key.callId); + } + }; + /// What a cancel is judged against: the facts the execute was admitted on. + struct CancelTarget { + std::shared_ptr<::core::async::StopSource> stop; + std::string principal; + std::string modelType; + std::string actionType; + std::uint64_t modelId{0}; + std::string owner; + }; + // Admitted Task executes on a connection scope that have not replied yet, + // filed by `fileCancellable` and withdrawn by `callFinished` or the cancel + // that stops them. + std::unordered_map _cancellable; std::uint64_t _nextId{0}; detail::OpaqueIdGenerator _idGen; // Admitted executes whose reply has not been sent: incremented at diff --git a/include/morph/core/wire.hpp b/include/morph/core/wire.hpp index f11600541..71e14a530 100644 --- a/include/morph/core/wire.hpp +++ b/include/morph/core/wire.hpp @@ -12,6 +12,7 @@ #include #include #include +#include #include "../session/session.hpp" #include "profiler.hpp" @@ -74,6 +75,12 @@ inline constexpr std::uint32_t kProtocolVersion = 1; /// see `makeHello()` and `interpretHelloReply()`. A peer /// predating negotiation answers `err "unknown envelope kind: /// hello"`, which is how a legacy server is detected. +/// - `"cancel"` — client asks the server to stop one of its own in-flight +/// `execute` calls. Uses `cancelCallId` (the execute's +/// `callId`) and `session`; its own `callId` correlates the +/// `ok` reply, which is the same whatever became of the call. +/// Sent only to a server whose `hello` reply advertised +/// `kCapabilityCancel`; see `makeCancel()`. struct Envelope { /// @brief Discriminator — see class docstring for valid values. std::string kind; @@ -143,8 +150,25 @@ struct Envelope { /// false`). Populated by `makeHello()` on `"hello"`; not otherwise /// inspected by other `kind`s today. std::uint32_t protocolVersion = 0; + + /// @brief The `callId` of the `execute` a `"cancel"` asks the server to stop. + /// + /// Carried in its own field rather than in `callId` because a `cancel` is + /// a request with a reply of its own: were the target's id in `callId`, the + /// server's `ok` to the cancel would be matched by the client against the + /// still-pending `execute` and settle it with an empty result. `0` names no + /// call. Ignored on every other kind. + std::uint64_t cancelCallId = 0; }; +/// @brief The capability a server's `"hello"` reply lists when it honours +/// `"cancel"` envelopes. +/// +/// A client sends `"cancel"` only to a server that advertised this; a server +/// that predates it never sees one. See "Cancelling a call" in +/// docs/spec/core/wire.md. +inline constexpr std::string_view kCapabilityCancel = "cancel"; + /// @brief Inclusive protocol-version range a server advertises in reply to /// `"hello"`. /// @@ -156,6 +180,13 @@ struct ProtocolRange { /// @brief Newest protocol version the server accepts. std::uint32_t max = kProtocolVersion; + + /// @brief Optional envelope kinds the server honours, beyond those every + /// server of this protocol version does (e.g. `kCapabilityCancel`). + /// + /// Additive: a reply from a server that predates the list has no such key, + /// and decodes to an empty list — "nothing optional is supported". + std::vector capabilities; }; /// @brief Builds a `register` envelope. @@ -265,6 +296,21 @@ inline Envelope makeSchemas(std::string typeId) { return env; } +/// @brief Builds a `cancel` envelope asking the server to stop the in-flight +/// `execute` filed under @p targetCallId on the sender's connection. +/// +/// The caller still assigns the envelope's own `callId` and `session` — the +/// same session the `execute` carried, since the server honours a cancel only +/// from the principal and connection that made the call. +/// @param targetCallId `callId` of the `execute` to stop. +/// @return The envelope, with `cancelCallId` set. +inline Envelope makeCancel(uint64_t targetCallId) { + Envelope env; + env.kind = "cancel"; + env.cancelCallId = targetCallId; + return env; +} + /// @brief Builds a `deregister` envelope. inline Envelope makeDeregister(uint64_t modelId) { Envelope env; @@ -398,9 +444,9 @@ struct EscapingWriteOpts : glz::opts { /// The general form of the rule, for anyone adding a member: **elide payload, /// never elide identity.** If a peer has to know the field's value to decide /// where the message goes, it is written even at its default. -inline constexpr std::array kOmittableEnvelopeKeys{ - "typeId", "contextKey", "primary", "shared", "modelId", "modelType", - "actionType", "body", "message", "session", "protocolVersion"}; +inline constexpr std::array kOmittableEnvelopeKeys{ + "typeId", "contextKey", "primary", "shared", "modelId", "modelType", + "actionType", "body", "message", "session", "protocolVersion", "cancelCallId"}; /// @brief Whether a session context carries nothing worth sending. /// @@ -431,7 +477,7 @@ inline constexpr std::array kOmittableEnvelopeKeys{ // exclude-write reports `unknown_key` for one that is not, which would turn // a member rename into a throw from *every* `encode` call at runtime. This // makes it a compile error instead. - static_assert(glz::reflect::size == 13, + static_assert(glz::reflect::size == 14, "Envelope gained or lost a member: decide whether it is omittable when default " "and update kOmittableEnvelopeKeys. A new member left off that list is always " "written, which is merely the old behaviour; one added with a wrong spelling is " @@ -482,6 +528,9 @@ inline constexpr std::array kOmittableEnvelopeKeys{ if (env.protocolVersion == 0) { omit("protocolVersion"); } + if (env.cancelCallId == 0) { + omit("cancelCallId"); + } return std::span{out.data(), count}; } @@ -742,4 +791,27 @@ inline ProtocolNegotiationResult interpretHelloReply(const Envelope& reply) { (reply.message.empty() ? std::string{"malformed reply"} : reply.message)); } +/// @brief Whether a reply to `"hello"` advertises @p capability. +/// +/// Reads the `ProtocolRange` in an `"ok"` reply's `body`, ignoring keys it does +/// not know. Anything else — an `"err"`, a legacy peer's refusal, a body that +/// does not parse, a server that predates the capability list — advertises +/// nothing, so a client falls back to not using the capability rather than to +/// sending a kind the server may not understand. +/// @param reply Decoded reply to a `"hello"` envelope. +/// @param capability Capability to look for, e.g. `kCapabilityCancel`. +/// @return `true` only when the reply is `"ok"` and its capability list names +/// @p capability. +[[nodiscard]] inline bool helloAdvertises(const Envelope& reply, std::string_view capability) { + if (reply.kind != "ok") { + return false; + } + ProtocolRange range; + static constexpr glz::opts kLenient{.null_terminated = false, .error_on_unknown_keys = false}; + if (glz::read(range, reply.body)) { + return false; + } + return std::ranges::find(range.capabilities, capability) != range.capabilities.end(); +} + } // namespace morph::wire diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 7d93f8955..52fdfae9b 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -55,6 +55,7 @@ add_executable(morph_tests test_remote_extra.cpp test_remote_server_strand.cpp test_remote_connection_scope.cpp + test_remote_cancel.cpp test_remote_step_interleaving.cpp test_remote_execute_ordering.cpp test_action_validation.cpp diff --git a/tests/test_remote_cancel.cpp b/tests/test_remote_cancel.cpp new file mode 100644 index 000000000..1814ee62c --- /dev/null +++ b/tests/test_remote_cancel.cpp @@ -0,0 +1,386 @@ +// SPDX-License-Identifier: Apache-2.0 +// +// The server side of a wire `cancel`: RemoteServer maps it onto the stop +// source of the execute it names, for the connection and principal that made +// the call and nobody else, and answers every cancel with the same `ok`. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "test_support.hpp" + +using morph::testing::WaitReply; +using morph::testing::waitUntil; + +// A named namespace, not an anonymous one: ActionDispatcher::registerAction +// files each action's schema, and glaze's reflection cannot name a type with +// internal linkage under MSVC (C7631). +namespace remote_cancel_test { + +struct Park { + int tag = 0; +}; + +/// What the handler reports back. The server constructs the model, so the +/// test reaches it through a global. +struct Probe { + std::atomic started{0}; + std::atomic stopped{0}; + std::atomic finished{0}; + std::unique_ptr> queue; +}; + +inline Probe& probe() { + static Probe instance; + return instance; +} + +/// A Task handler that blocks until an item arrives on the probe's queue, or +/// until it is stopped: only a push or a stop can end it. +struct ParkModel { + // NOLINTNEXTLINE(readability-convert-member-functions-to-static) + core::async::Task execute(Park action) { + probe().started.fetch_add(1); + try { + auto const item = co_await probe().queue->pop(); + probe().finished.fetch_add(1); + co_return item.value_or(0) + action.tag; + } catch (const core::async::OperationCancelled&) { + probe().stopped.fetch_add(1); + probe().finished.fetch_add(1); + throw; + } + } +}; + +} // namespace remote_cancel_test + +using remote_cancel_test::Park; +using remote_cancel_test::ParkModel; +using remote_cancel_test::probe; + +template <> +struct morph::model::ModelTraits { + static constexpr std::string_view typeId() { return "RC_Park"; } +}; +template <> +struct morph::model::ActionTraits { + using Result = int; + static constexpr std::string_view typeId() { return "RC_ParkAction"; } + static std::string toJson(const Park&) { return "{}"; } + static Park fromJson(std::string_view) { return {}; } + static std::string resultToJson(const int& result) { return std::to_string(result); } + static int resultFromJson(std::string_view json) { return std::stoi(std::string{json}); } +}; + +namespace { + +/// Derives the principal from the token alone, so a claimed principal that +/// disagrees with the token is visible. Type-level `authorize` admits exactly +/// the callers it can authenticate; ownership admits an unowned instance to +/// everyone and an owned one to its owner. +struct TokenAuthorizer : morph::session::IAuthorizer { + [[nodiscard]] static std::optional principalFor(std::string_view token) { + if (token == "tok-alice") { + return std::string{"alice"}; + } + if (token == "tok-bob") { + return std::string{"bob"}; + } + return std::nullopt; + } + [[nodiscard]] bool authorize(const morph::session::Context& ctx, std::string_view, + std::string_view) const override { + return principalFor(ctx.token).has_value(); + } + [[nodiscard]] std::optional authenticate(const morph::session::Context& ctx) const override { + return principalFor(ctx.token); + } + [[nodiscard]] bool authorizeInstance(const morph::session::Context& ctx, std::string_view, std::string_view, + std::uint64_t, std::string_view ownerPrincipal) const override { + return ownerPrincipal.empty() || ownerPrincipal == ctx.principal; + } +}; + +/// Who a message claims to come from. +struct Caller { + std::string token; + std::string claimed; +}; + +/// One server with the parking model, and a queue whose consumer is resumed on +/// the executor it parked on. +struct Rig { + // Declared before the executors, so they outlive every task the pool runs. + morph::model::detail::ActionDispatcher dispatcher; + morph::model::detail::ModelRegistryFactory registry; + core::async::ThreadPoolExecutor foreign{1}; + morph::exec::ThreadPoolExecutor pool{2}; + std::shared_ptr server; + + explicit Rig(std::shared_ptr authorizer = nullptr) { + registry.registerModel("RC_Park"); + dispatcher.registerAction("RC_Park", "RC_ParkAction"); + server = std::make_shared(pool, std::move(authorizer), dispatcher, registry); + probe().started = 0; + probe().stopped = 0; + probe().finished = 0; + probe().queue = std::make_unique>(foreign, core::async::AsyncQueueOptions{}); + } + Rig(const Rig&) = delete; + Rig& operator=(const Rig&) = delete; + Rig(Rig&&) = delete; + Rig& operator=(Rig&&) = delete; + + // Unparks any handler still waiting -- a case that failed leaves one -- and + // waits for it to end before the queue goes: a queue must not die under a + // parked consumer. + ~Rig() { + probe().queue->close(); + static_cast(waitUntil([] { return probe().finished.load() == probe().started.load(); })); + probe().queue.reset(); + } + + [[nodiscard]] morph::wire::Envelope ask(const morph::wire::Envelope& env, morph::backend::ConnectionId cid) const { + WaitReply reply; + server->handle(morph::wire::encode(env), std::ref(reply), cid); + REQUIRE(reply.await()); + return reply.env; + } + + [[nodiscard]] std::uint64_t registerOn(morph::backend::ConnectionId cid, const Caller& caller) const { + auto env = morph::wire::makeRegister("RC_Park"); + env.session.token = caller.token; + env.session.principal = caller.claimed; + auto const reply = ask(env, cid); + REQUIRE(reply.kind == "ok"); + return reply.modelId; + } +}; + +/// Where an execute is sent, and under which `callId`. +struct Target { + morph::backend::ConnectionId cid{0}; + std::uint64_t modelId{0}; + std::uint64_t callId{0}; +}; + +morph::wire::Envelope parkCall(const Target& target, const Caller& caller) { + morph::wire::Envelope env; + env.kind = "execute"; + env.callId = target.callId; + env.modelId = target.modelId; + env.modelType = "RC_Park"; + env.actionType = "RC_ParkAction"; + env.body = "{}"; + env.session.token = caller.token; + env.session.principal = caller.claimed; + return env; +} + +morph::wire::Envelope cancelOf(std::uint64_t targetCallId, const Caller& caller) { + auto env = morph::wire::makeCancel(targetCallId); + env.callId = targetCallId + 1000U; + env.session.token = caller.token; + env.session.principal = caller.claimed; + return env; +} + +/// Pushes one item and waits for the parked call's reply. A call nothing +/// stopped answers `ok` with the item; a stopped one has already unwound, so +/// the item reaches nobody and the reply is the stop's `err`. +morph::wire::Envelope releaseAndAwait(WaitReply& call) { + static_cast(probe().queue->push(41)); + REQUIRE(call.await()); + return call.env; +} + +} // namespace + +TEST_CASE("RemoteServer: a cancel stops the Task handler of the call it names", "[remote][cancel]") { + Rig const rig; + auto const cid = rig.server->openConnection(); + auto const mid = rig.registerOn(cid, {}); + + WaitReply call; + rig.server->handle(morph::wire::encode(parkCall({.cid = cid, .modelId = mid, .callId = 7}, {})), std::ref(call), + cid); + REQUIRE(waitUntil([] { return probe().started.load() == 1; })); + + auto const cancelled = rig.ask(cancelOf(7, {}), cid); + // The cancel's own reply, under its own callId: were it addressed to the + // execute, the client would settle the call with it. + CHECK(cancelled.kind == "ok"); + CHECK(cancelled.callId == 1007U); + + REQUIRE(call.await()); + CHECK(call.env.kind == "err"); + CHECK(call.env.callId == 7U); + CHECK(probe().stopped.load() == 1); +} + +TEST_CASE("RemoteServer: a cancel from another connection has no effect on the call", "[remote][cancel]") { + Rig const rig; + auto const owner = rig.server->openConnection(); + auto const other = rig.server->openConnection(); + auto const mid = rig.registerOn(owner, {}); + + WaitReply call; + rig.server->handle(morph::wire::encode(parkCall({.cid = owner, .modelId = mid, .callId = 7}, {})), std::ref(call), + owner); + REQUIRE(waitUntil([] { return probe().started.load() == 1; })); + + // Same callId, same (empty) principal: only the connection differs. + auto const foreign = rig.ask(cancelOf(7, {}), other); + CHECK(foreign.kind == "ok"); + + auto const reply = releaseAndAwait(call); + CHECK(reply.kind == "ok"); + CHECK(reply.body == "41"); + CHECK(probe().stopped.load() == 0); +} + +TEST_CASE("RemoteServer: a cancel keys on the verified principal, not the claimed one", "[remote][cancel]") { + // An unowned instance two principals both use over one connection -- a + // gateway multiplexing its users -- so `authorizeInstance` admits both, + // and the principal recorded at the execute is all that tells them apart. + Rig const rig{std::make_shared()}; + auto const cid = rig.server->openConnection(); + auto const mid = rig.registerOn(cid, {.token = {}, .claimed = {}}); + Caller const alice{.token = "tok-alice", .claimed = "alice"}; + + WaitReply call; + rig.server->handle(morph::wire::encode(parkCall({.cid = cid, .modelId = mid, .callId = 7}, alice)), std::ref(call), + cid); + REQUIRE(waitUntil([] { return probe().started.load() == 1; })); + + SECTION("bob's token claiming to be alice has no effect") { + auto const forged = rig.ask(cancelOf(7, {.token = "tok-bob", .claimed = "alice"}), cid); + CHECK(forged.kind == "ok"); + + auto const reply = releaseAndAwait(call); + CHECK(reply.kind == "ok"); + CHECK(probe().stopped.load() == 0); + } + + SECTION("a claim of alice with no token has no effect") { + auto const tokenless = rig.ask(cancelOf(7, {.token = {}, .claimed = "alice"}), cid); + CHECK(tokenless.kind == "ok"); + + auto const reply = releaseAndAwait(call); + CHECK(reply.kind == "ok"); + CHECK(probe().stopped.load() == 0); + } + + SECTION("control: alice's own token stops it, whatever she claims") { + auto const own = rig.ask(cancelOf(7, {.token = "tok-alice", .claimed = "mallory"}), cid); + CHECK(own.kind == "ok"); + + REQUIRE(call.await()); + CHECK(call.env.kind == "err"); + CHECK(probe().stopped.load() == 1); + } +} + +TEST_CASE("RemoteServer: on an owned instance only the owner's cancel stops the call", "[remote][cancel]") { + // Alice owns the instance. A cancel carrying alice's verified principal + // stops it; nothing else reaches it -- not even bob on alice's connection. + Rig const rig{std::make_shared()}; + auto const cid = rig.server->openConnection(); + Caller const alice{.token = "tok-alice", .claimed = "alice"}; + auto const mid = rig.registerOn(cid, alice); + + WaitReply call; + rig.server->handle(morph::wire::encode(parkCall({.cid = cid, .modelId = mid, .callId = 9}, alice)), std::ref(call), + cid); + REQUIRE(waitUntil([] { return probe().started.load() == 1; })); + + auto const bob = rig.ask(cancelOf(9, {.token = "tok-bob", .claimed = "bob"}), cid); + CHECK(bob.kind == "ok"); + auto const own = rig.ask(cancelOf(9, alice), cid); + CHECK(own.kind == "ok"); + + REQUIRE(call.await()); + CHECK(call.env.kind == "err"); + CHECK(probe().stopped.load() == 1); +} + +TEST_CASE("RemoteServer: a cancel for an unknown, finished or unscoped call is a no-op ok", "[remote][cancel]") { + Rig const rig; + auto const cid = rig.server->openConnection(); + auto const mid = rig.registerOn(cid, {}); + + SECTION("unknown") { + auto const reply = rig.ask(cancelOf(12345, {}), cid); + CHECK(reply.kind == "ok"); + CHECK(reply.callId == 13345U); + } + + SECTION("finished") { + WaitReply call; + rig.server->handle(morph::wire::encode(parkCall({.cid = cid, .modelId = mid, .callId = 3}, {})), + std::ref(call), cid); + REQUIRE(waitUntil([] { return probe().started.load() == 1; })); + CHECK(releaseAndAwait(call).kind == "ok"); + + auto const reply = rig.ask(cancelOf(3, {}), cid); + CHECK(reply.kind == "ok"); + } + + SECTION("unscoped: a message with no connection names no call") { + auto const unscoped = rig.registerOn(0, {}); + WaitReply call; + rig.server->handle(morph::wire::encode(parkCall({.cid = 0, .modelId = unscoped, .callId = 5}, {})), + std::ref(call), 0); + REQUIRE(waitUntil([] { return probe().started.load() == 1; })); + + CHECK(rig.ask(cancelOf(5, {}), 0).kind == "ok"); + + auto const reply = releaseAndAwait(call); + CHECK(reply.kind == "ok"); + CHECK(probe().stopped.load() == 0); + } +} + +TEST_CASE("RemoteServer: hello advertises the cancel capability", "[remote][cancel][protocol]") { + Rig const rig; + auto const reply = rig.ask(morph::wire::makeHello(), 0); + REQUIRE(reply.kind == "ok"); + CHECK(morph::wire::helloAdvertises(reply, morph::wire::kCapabilityCancel)); +} + +TEST_CASE("wire::helloAdvertises: a server that predates the capability list advertises nothing", + "[wire][cancel][protocol]") { + CHECK_FALSE( + morph::wire::helloAdvertises(morph::wire::makeOk(1, R"({"min":1,"max":1})"), morph::wire::kCapabilityCancel)); + CHECK_FALSE(morph::wire::helloAdvertises(morph::wire::makeErr("unknown envelope kind: hello", 1), + morph::wire::kCapabilityCancel)); + CHECK_FALSE(morph::wire::helloAdvertises(morph::wire::makeOk(1, "not json"), morph::wire::kCapabilityCancel)); + CHECK(morph::wire::helloAdvertises(morph::wire::makeOk(1, R"({"min":1,"max":1,"capabilities":["cancel"]})"), + morph::wire::kCapabilityCancel)); +} + +TEST_CASE("wire::makeCancel carries the target in cancelCallId and round-trips", "[wire][cancel]") { + auto env = morph::wire::makeCancel(77); + env.callId = 78; + auto const decoded = morph::wire::decode(morph::wire::encode(env)); + CHECK(decoded.kind == "cancel"); + CHECK(decoded.callId == 78U); + CHECK(decoded.cancelCallId == 77U); +} diff --git a/tests/test_wire_omitted_fields.cpp b/tests/test_wire_omitted_fields.cpp index 61a0936d3..2da789851 100644 --- a/tests/test_wire_omitted_fields.cpp +++ b/tests/test_wire_omitted_fields.cpp @@ -58,6 +58,7 @@ morph::wire::Envelope fullyPopulated() { env.session.locale = "fr-FR"; env.session.metadata["flag"] = "on"; env.protocolVersion = 7U; + env.cancelCallId = 11U; return env; } @@ -79,6 +80,7 @@ void requireEqual(const morph::wire::Envelope& lhs, const morph::wire::Envelope& CHECK(lhs.session.locale == rhs.session.locale); CHECK(lhs.session.metadata == rhs.session.metadata); CHECK(lhs.protocolVersion == rhs.protocolVersion); + CHECK(lhs.cancelCallId == rhs.cancelCallId); } } // namespace From c526c2cbe14ff74f6ad12a7d5c64713d83403d59 Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Sun, 4 Oct 2026 12:40:00 +0200 Subject: [PATCH 2/2] net, qt: SocketBackend and QtWebSocketBackend send cancel when a call's stop is requested (#865) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both backends learn from the opt-in negotiateProtocolVersion whether the server honours "cancel" (SocketBackend gains the verb), re-send hello after every reconnect and forget the capability on every disconnect. Only then do they send `cancel {cancelCallId}`, fire-and-forget under a fresh callId: - when a call's stop is requested (an execute deadline, or any holder of ActionCall::stopSource) — a stop callback registered once the call has its id and its frame is queued, which hands the cancel to the backend's owner; - for every execute cancelPending sweeps (~Bridge, switchBackend). The cancel carries the execute's own session, since the server honours it only under the execute's verified principal. Teardown lets the cancels out: SocketBackend's close is a close-after-flush on the loop, and ~QtWebSocketBackend flushes before its abort. Both discarded the frames ~Bridge's cancelPending had just queued. SocketBackend::cancelPending is posted to the I/O loop (G0 on return); the backend spec states why a cancel queued behind earlier loop work is acceptable: it can never overtake its execute, the delay is only to the server's handler, and the caller's own completion is settled in that task. Tests over real loopbacks (SocketServer + SocketBackend, QtWebSocketServer + QtWebSocketBackend): a client deadline, and destroying the Bridge, each make the server's Task handler observe the stop; a client that did not negotiate, or negotiated with a legacy server, sends none. Co-Authored-By: Claude Opus 5.5 --- docs/spec/core/backend.md | 105 ++++++++++- docs/spec/core/wire.md | 5 + include/morph/net/socket_backend.hpp | 188 +++++++++++++++++++- include/morph/qt/qt_websocket_backend.hpp | 53 +++++- scripts/branch_partial_allowlist.json | 2 +- src/qt/qt_websocket_backend.cpp | 87 ++++++++- tests/net/test_socket_backend.cpp | 206 ++++++++++++++++++++++ tests/qt/test_qt_websocket.cpp | 121 +++++++++++++ 8 files changed, 751 insertions(+), 16 deletions(-) diff --git a/docs/spec/core/backend.md b/docs/spec/core/backend.md index bc1508ab1..1641cd04d 100644 --- a/docs/spec/core/backend.md +++ b/docs/spec/core/backend.md @@ -1218,7 +1218,9 @@ socket connects and before any `bindModel`/`execute`, but nothing enforces that ordering and nothing sends it automatically. Rejects with `std::runtime_error` if the server explicitly rejects the version or the socket is not connected, and with `DisconnectedError` if it drops before the reply. See -[wire.md](wire.md#protocol-version-negotiation). +[wire.md](wire.md#protocol-version-negotiation). The reply also tells the +backend whether the server honours `"cancel"`; see +[Cancelling a remote call on the server](#cancelling-a-remote-call-on-the-server). **TLS.** Pass a `QSslConfiguration` to enable `wss://`. Build it with `tlsVerifyingConfig()` (CA-verified, the recommended production default) or @@ -1558,6 +1560,98 @@ listening socket's `O_NONBLOCK` onto the sockets `accept(2)` returns (POSIX permits this; Linux does not do it), and a blocking reader of such a socket fails on `EAGAIN` before its peer has written. +## Cancelling a remote call on the server + +`SocketBackend` and `QtWebSocketBackend` tell the server when they abandon an +`execute`, by sending `cancel {cancelCallId}` (see +[wire.md](wire.md#cancelling-a-call)), so a server-side Task handler stops +instead of running on for a reply nobody reads. Each sends one only to a +server that advertised `"cancel"` in its `"hello"` reply, so each sends none +until the application has called `negotiateProtocolVersion` — opt-in, like +the handshake itself. Once called, the backend re-sends `"hello"` after every +reconnect and forgets the capability on every disconnect: a new connection may +reach a different server. + +A call is cancelled on the server when: + +- **Its stop is requested** — an execute deadline (`Bridge::setExecuteDeadline`) + or any other holder of the call's `ActionCall::stopSource`. The backend + registers a stop callback on that source once the call has its `callId` and + its frame is queued. The callback runs on whichever thread requested the + stop, so it touches no table: `SocketBackend`'s posts `requestCancel(callId)` + to the I/O loop; `QtWebSocketBackend`'s sets the call's flag and starts a + zero-interval single-shot wake timer through the member-name + `QMetaObject::invokeMethod(&timer, "start", Qt::QueuedConnection)` — which + allocates nothing, unlike the functor overload — and the timer's drain, on + the Qt thread, sends a cancel for each flagged call. Either way the owner + sends the cancel only if the call is still waiting for its reply, and the + pending table stays the owner's alone, with no lock. A stop after the reply sends + nothing: the callback is deregistered with the pending entry. A call with no + stop source — `Bridge` gives one only to a call whose handler is a Task, + the one kind a server can stop — never registers one. +- **`cancelPending` sweeps it** — `~Bridge` and `Bridge::switchBackend` call it. + Every drained execute is cancelled, with or without a stop source: in a + client-only build `Bridge` cannot tell a Task handler from an ordinary one, + and the server answers a cancel for an unstoppable call with the same + no-op `ok`. + +The cancel carries the session the `execute` carried (kept with the pending +entry for a stoppable call; the backend's control session otherwise), since +the server honours it only under the execute's verified principal. It is sent +fire-and-forget under a fresh `callId` filed nowhere, so its `ok` is dropped +by the reply router. + +**Ordering, and why `SocketBackend::cancelPending` being posted is acceptable.** +`SocketBackend::cancelPending` is posted to the I/O loop (**G0** on return), so +the cancels it sends sit behind whatever the loop already had queued. That is +acceptable, for three reasons. The cancel can never overtake the execute it +names: the execute was posted to the same loop before it, the loop runs its +tasks in order and writes one connection's frames in order, and the server +handles one connection's messages in order, so the execute is always admitted +before its cancel is read. What waits is only the server's handler, which +runs on for at most as long as the loop takes to drain its queue — work that +never blocks — and which already ran to its end before cancels existed. And +the caller's own guarantee does not depend on it: the calls are settled with +the cancel's exception in the same loop task, before any reply to them can be +delivered. A stop-triggered cancel is posted in the same way, from the stopping +thread, and is ordered the same. + +**Teardown.** `~Bridge` calls `cancelPending` and then destroys the backend, +so the cancels are queued just before the backend closes its socket. Each +backend's close therefore lets queued frames out first: `SocketBackend`'s +close is a close-after-flush on the loop (the write still bounded by +`sendTimeout`), and `QtWebSocketBackend`'s destructor flushes its socket, +without blocking, before the abort. A close that discarded queued frames +would drop exactly these cancels; the `~Bridge` tests in both suites fail +when it does (measured on both backends). `QtWebSocketBackend` sends nothing from its own +destructor's sweep — its socket is already aborted there — so a backend +destroyed without its owner's `cancelPending` cancels nothing. + +### Cancellation-policy rows for the remote backends + +The policy table in +[concurrency_and_lifetimes.md](../concurrency_and_lifetimes.md#every-cancel-verb-measured) +marks the remote backends' rows read, not measured, and gives "the server +keeps executing" for the work they abandon. What those rows are, and which +parts are measured, is: + +- `SocketBackend::cancelPending` — **G0** on return; **G2** once the I/O loop + runs it (unchanged). Work: a Task handler on a server that advertised + `"cancel"` is asked to stop; otherwise the server keeps executing. + **Measured** over a real loopback (`SocketServer` with `SocketBackend`) for + the stop, through `~Bridge`. +- `QtWebSocketBackend::cancelPending` — **G2** (unchanged). Work: as above. + **Measured** in the Qt suite against a real `QtWebSocketServer`, through + `~Bridge`. +- `Bridge::setExecuteDeadline` over either backend — the deadline's stop + reaches the server's Task handler through a cancel. **Measured** for both + over a real loopback. + +The G-levels of these verbs are unchanged and are still read, not measured: +the tests above measure the work column only. The measuring tests are in +`tests/net/test_socket_backend.cpp` and `tests/qt/test_qt_websocket.cpp` +(`[cancel]`). + ## Lifetime & ownership The backends hold *references*, not owning pointers, to the resources they run @@ -1861,12 +1955,12 @@ server: each call is a loop task, so the loop serialises them. | `promoteModel(request, cbExec)` | Sends `assign` through the same path. An empty `primary` or zero `mid` resolves with `request.mid` without sending. | | `instances(typeId, cbExec)` | Sends `instances` through the same path; the reply's body is decoded into the key list. Rejects with `"disconnected"` when not connected. | | `waitForConnected(timeoutMs = 5000)` | Pumps a local `QEventLoop` until connected or timeout; returns `_connected`. Not for a WASM main thread. | -| `negotiateProtocolVersion(replyExec)` | Opt-in: sends `hello` through the same path and settles the returned `Completion` with `wire::interpretHelloReply`'s classification. Rejects on an explicit version rejection, when not connected, or on a drop. | +| `negotiateProtocolVersion(replyExec)` | Opt-in: sends `hello` through the same path and settles the returned `Completion` with `wire::interpretHelloReply`'s classification; records whether the server advertised `"cancel"`, and re-sends `hello` after every reconnect. Rejects on an explicit version rejection, when not connected, or on a drop. | | `registerModel(typeId, factory)`, `assignPrimary(...)`, `listInstances(typeId)` | Throw `std::logic_error`: each would have to wait for a reply. `registerModelWithContext` is `IBackend`'s default, which forwards to `registerModel`. | | `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, recorded in `_pendingDeregisters` so `onTextMessage` recognises the reply and drops it. | | `execute(mid, call, cbExec)` | Assigns a `callId`, sends `execute`, returns a `Completion`. Immediate `DisconnectedError` if not connected. | | `notifyBackendChanged()` | No-op. | -| `cancelPending(exc)` | On the socket's thread, takes `_pending`, `_pendingControl` and `_queuedRegistrations` out, then delivers `exc` to each — the exception itself, so a request rejected by a dropped socket carries the same `DisconnectedError` an `execute` does. | +| `cancelPending(exc)` | On the socket's thread, takes `_pending`, `_pendingControl` and `_queuedRegistrations` out, sends a `cancel` for each `execute` when connected to a server that advertised `"cancel"`, then delivers `exc` to each — the exception itself, so a request rejected by a dropped socket carries the same `DisconnectedError` an `execute` does. | | `setReconnectHandler(handler)` | Stores the handler; invoked on the Qt thread after every *subsequent* connect. `nullptr` clears. | | `setConnectHandler(handler)` | Stores the handler; invoked on the Qt thread after every successful connect, first included. `nullptr` clears. | | `setDisconnectHandler(handler)` | Stores the handler; invoked on the Qt thread whenever the socket drops, before reconnect scheduling. `nullptr` clears. | @@ -1917,7 +2011,7 @@ not a behavior change to the existing loopback-only default. |---|---| | `SocketBackend(loop, serverUrl, cfg = Config{})` | Parses `serverUrl` (`ws://` only — throws immediately on `wss://`) and posts the first connection attempt to `loop`, an `exec::IoLoop` that must outlive the backend. | | `explicit SocketBackend(serverUrl, cfg = Config{})` | The same, on a private `IoLoop` the backend owns. The URL is parsed before that loop is started. | -| `~SocketBackend()` | Runs its close on the loop — inline on the loop's own thread, posted and waited for otherwise: closes the connection, retires the backoff timer, rejects every pending call — a waiting synchronous verb's included — with `DisconnectedError`. Never waits for a dial in progress. | +| `~SocketBackend()` | Runs its close on the loop — inline on the loop's own thread, posted and waited for otherwise: closes the connection, retires the backoff timer, rejects every pending call — a waiting synchronous verb's included — with `DisconnectedError`. Frames already queued — the cancels and deregisters of a `cancelPending` just before — are written before the socket closes, bounded by `sendTimeout`. Never waits for a dial in progress. | | `waitForConnected(timeout = 5000ms)` | Posts a waiter the loop releases on the next completed connection, and blocks the calling thread on it until then or the timeout; returns the current connected state. On the loop's own thread it answers at once. 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)` | Forwards to `registerModelWithContext` with an empty `contextKey`; `factory` ignored. | | `registerModelWithContext(typeId, factory, contextKey)` | Synchronous: posts `register` carrying `contextKey` and waits on a future for the loop to settle the reply, so the server's `LogProvider` is consulted for a private registration exactly as it is for a shared one. `factory` ignored. Throws on `err` reply or disconnect, and on the loop's own thread. Any number may be in flight, matched by `callId`. | @@ -1926,7 +2020,8 @@ not a behavior change to the existing loopback-only default. | `deregisterModel(mid)` | **Fire-and-forget** — posted only if connected, does not wait for the ack. The loop gives it a non-zero `callId` from the same counter `execute` uses, so its unawaited `ok` is never taken for another call's reply. Needs no pending-id bookkeeping of its own: the reply router already drops a non-zero `callId` that is not pending. | | `execute(mid, call, cbExec)` | Serialises the action on the calling thread and posts the envelope; the loop assigns a `callId`, files the pending record and writes the frame. Returns a `Completion`, already rejected with `DisconnectedError` if not connected. Callable from any thread; any number may be in flight. | | `notifyBackendChanged()` | No-op. | -| `cancelPending(exc)` | Posted: the loop drains **both** pending tables — the `execute` calls and the `bindModel`/`promoteModel` control calls — and delivers `exc` to each state. Covers every call issued before it. | +| `cancelPending(exc)` | Posted: the loop drains **both** pending tables — the `execute` calls and the `bindModel`/`promoteModel` control calls — and delivers `exc` to each state. Covers every call issued before it. When the server advertised `"cancel"`, sends one for each drained `execute` first. | +| `negotiateProtocolVersion(replyExec)` | Opt-in: posts a `hello` on the callId-multiplexed path and settles the returned `Completion` with `wire::interpretHelloReply`'s classification; records whether the server advertised `"cancel"`, and re-sends `hello` after every reconnect. Rejects on an explicit version rejection, and with `DisconnectedError` when the socket is down or drops. | | `setReconnectHandler(handler, exec)` | Posted: the loop stores the handler and its executor, and after every *subsequent* connect posts the handler to `exec` — never runs it on the loop. `nullptr` clears. | | `setSession(session)` | Posted: the loop stores the session it stamps onto every control envelope it builds afterwards. | diff --git a/docs/spec/core/wire.md b/docs/spec/core/wire.md index 2e5b683ca..d772bf4a1 100644 --- a/docs/spec/core/wire.md +++ b/docs/spec/core/wire.md @@ -521,6 +521,11 @@ cancel naming it is a no-op like any other. A call still queued behind another on its instance's action gate is stopped before it starts: its handler's token is already stopped when it runs. +Two clients send it: `SocketBackend` and `QtWebSocketBackend`, after their +opt-in `negotiateProtocolVersion` has found the capability — see +[backend.md](backend.md#cancelling-a-remote-call-on-the-server) for when each +sends one. + An unscoped message (`ConnectionId` `0`, the two-argument `handle()`) names no call: the server files no unscoped call as cancellable, because unscoped callers share one id space and could not be told apart. diff --git a/include/morph/net/socket_backend.hpp b/include/morph/net/socket_backend.hpp index f6d3716bd..ddfbe81e9 100644 --- a/include/morph/net/socket_backend.hpp +++ b/include/morph/net/socket_backend.hpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -316,6 +317,32 @@ class SocketBackend : public ::morph::backend::detail::IBackend { return keys; } + /// @brief Sends a `"hello"` envelope and classifies the server's reply. + /// + /// Opt-in, as on the other backends: a backend that never calls it sends + /// no `"hello"`. Calling it also learns whether the server honours + /// `"cancel"` (`wire::kCapabilityCancel`), which is the only way this + /// backend ever sends one; and it re-sends `"hello"` on every reconnect, + /// since a new connection may reach a different server. + /// @param replyExec Executor the answer is delivered on. Borrowed: it must + /// outlive the returned `Completion`. + /// @return A `Completion` resolved with `Negotiated` or `LegacyPeer`; + /// rejected with a `std::runtime_error` if the server refuses the + /// version, and with `backend::DisconnectedError` if the socket is + /// down or drops before the reply. + [[nodiscard]] ::morph::async::Completion<::morph::wire::ProtocolNegotiationResult> negotiateProtocolVersion( + ::morph::exec::IExecutor& replyExec MORPH_LIFETIMEBOUND) { + using Result = ::morph::wire::ProtocolNegotiationResult; + auto const state = std::make_shared<::morph::async::detail::CompletionState>(); + ::morph::async::Completion comp{state, &replyExec}; + _loop->post([core = _core, state] { + core->note("SocketBackend::negotiateProtocolVersion"); + core->negotiate = true; + core->fileHello(state); + }); + return comp; + } + /// @brief Sends a `deregister` message fire-and-forget (does not wait for a reply). /// /// Posted to the loop, which gives it a real, non-zero `callId` drawn from @@ -364,8 +391,14 @@ class SocketBackend : public ::morph::backend::detail::IBackend { env.body = call.serializeBody(); env.session = std::move(call.session); - _loop->post([core = _core, env, - pending = PendingExecute{.state = compState, .deserialize = call.deserializeResult}]() mutable { + PendingExecute pending{.state = compState, .deserialize = call.deserializeResult}; + if (call.stopSource) { + // Kept for the cancel a stop sends: the server honours it only + // under the principal the execute was admitted with. + pending.stop = call.stopSource; + pending.session = env.session; + } + _loop->post([core = _core, env, pending = std::move(pending)] mutable { core->fileExecute(std::move(env), std::move(pending)); }); return comp; @@ -380,13 +413,15 @@ class SocketBackend : public ::morph::backend::detail::IBackend { /// runs posts in order. Covers both in-flight tables: the `execute` calls /// and the `bindModel`/`promoteModel` control calls. 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. + /// only arrive on a connection that is gone. When the server advertised + /// `"cancel"`, each abandoned `execute` is also cancelled on the server, + /// from the loop, behind whatever the loop had queued before. /// @param exc Exception delivered to every pending completion's error sink. void cancelPending(const std::exception_ptr& exc) override { _loop->post([core = _core, exc] { ::morph::exec::detail::noteOwner("SocketBackend::cancelPending", core->loop.loop(), core->loop.runningHere()); - core->cancelAll(exc); + core->cancelAll(exc, CancelOnServer::Yes); }); } @@ -485,9 +520,43 @@ class SocketBackend : public ::morph::backend::detail::IBackend { return std::move(*value); } + struct Core; + + /// Whether settling the pending calls also tells the server to stop them. + enum class CancelOnServer : std::uint8_t { No, Yes }; + + /// The stop callback a stoppable execute registers: hands the cancel to + /// the loop, which owns the socket and the pending table. Runs on whichever + /// thread requested the stop, so it only posts. + struct CancelOnStop { + std::weak_ptr core; + ::morph::exec::IoLoop* loop{nullptr}; + std::uint64_t callId{0}; + void operator()() const noexcept { + try { + loop->post([weak = core, target = callId] { + if (auto const live = weak.lock()) { + live->requestCancel(target); + } + }); + } catch (...) { // NOLINT(bugprone-empty-catch): a stop callback must not throw + // The loop could not take the task: the server is not told, + // and the call runs to its end there, as without a cancel. + } + } + }; + struct PendingExecute { std::shared_ptr<::morph::async::detail::CompletionState>> state; std::shared_ptr (*deserialize)(std::string_view json){nullptr}; + /// The call's stop source, or null for a call nothing can stop. + std::shared_ptr<::core::async::StopSource> stop; + /// The execute's session, for the cancel; empty without `stop`. + ::morph::session::Context session; + /// Registered on `stop` once the call has its id; deregistered when the + /// entry goes, so a stop after the reply sends nothing. Shared, not + /// unique, because the entry travels in a copyable loop task. + std::shared_ptr<::core::async::StopCallback> watch; }; /// @brief One in-flight control call issued through `bindModel`/`promoteModel`. @@ -510,6 +579,12 @@ class SocketBackend : public ::morph::backend::detail::IBackend { /// @brief State of a call answered with the reply's body /// (`instances`), or null. std::shared_ptr<::morph::async::detail::CompletionState> body; + /// @brief Whether this is a `"hello"`, whose reply says whether the + /// server honours `"cancel"`. + bool hello = false; + /// @brief State of a `negotiateProtocolVersion` call, or null — null + /// for the `"hello"` a reconnect re-sends. + std::shared_ptr<::morph::async::detail::CompletionState<::morph::wire::ProtocolNegotiationResult>> negotiation; /// @brief Rejects whichever state this call settles. /// @param failure The failure. @@ -520,6 +595,9 @@ class SocketBackend : public ::morph::backend::detail::IBackend { if (body) { body->setException(failure); } + if (negotiation) { + negotiation->setException(failure); + } } }; @@ -548,6 +626,18 @@ class SocketBackend : public ::morph::backend::detail::IBackend { /// @param pending Entry taken out of the control table. /// @param reply Decoded reply carrying the same `callId`. static void settleControl(const PendingControl& pending, const ::morph::wire::Envelope& reply) { + if (pending.hello) { + // A legacy peer's refusal is an answer, not a failure: classified, + // never prefixed with "hello failed". + if (pending.negotiation) { + try { + pending.negotiation->setValue(::morph::wire::interpretHelloReply(reply)); + } catch (...) { + pending.negotiation->setException(std::current_exception()); + } + } + return; + } if (reply.kind != "ok") { pending.reject(std::make_exception_ptr(std::runtime_error(pending.what + " failed: " + reply.message))); return; @@ -613,10 +703,68 @@ class SocketBackend : public ::morph::backend::detail::IBackend { pending.state->setException(std::current_exception()); return; } + auto const stop = pending.stop; executes.insert(callId, std::move(pending)); // A refused frame is left filed: the connection is closing, and // the disconnect that follows sweeps it with `DisconnectedError`. static_cast(send(std::move(frame))); + if (stop) { + // After the frame is queued, so a cancel never precedes its + // execute on the wire. A stop already requested runs the + // callback here, which posts the cancel behind this task. + auto watch = std::make_shared<::core::async::StopCallback>( + stop->get_token(), CancelOnStop{.core = weak_from_this(), .loop = &loop, .callId = callId}); + if (auto entry = executes.take(callId)) { + entry->watch = std::move(watch); + executes.insert(callId, std::move(*entry)); + } + } + } + + /// The stop of the call filed under @p callId was requested: tells the + /// server, if the call is still waiting for its reply and the server + /// advertised `"cancel"`. On the loop. + void requestCancel(std::uint64_t callId) { + note("SocketBackend::requestCancel"); + if (!cancelAdvertised) { + return; + } + // Taken and put back: the entry stays pending, its reply still + // settles it, and this is the table's one way to read an entry. + auto entry = executes.take(callId); + if (!entry) { + return; + } + sendCancel(callId, entry->stop ? entry->session : session); + executes.insert(callId, std::move(*entry)); + } + + /// Writes a fire-and-forget `cancel` for @p target. Its reply is + /// filed nowhere, so the router drops it. On the loop. + void sendCancel(std::uint64_t target, const ::morph::session::Context& callerSession) { + auto env = ::morph::wire::makeCancel(target); + env.callId = executes.nextCallId(); + env.session = callerSession; + try { + static_cast(send(::morph::net::detail::encodeWsFrame( + ::morph::net::detail::WsOpcode::kText, ::morph::wire::encode(env), /*mask=*/true))); + } catch (const std::exception&) { // NOLINT(bugprone-empty-catch) + // Best effort, like a deregister: without the cancel the + // server's handler runs to its end, as it did before cancels. + } + } + + /// Files a `"hello"`, settling @p negotiation (null for a reconnect's + /// re-send) from its reply. On the loop. + void fileHello( + std::shared_ptr<::morph::async::detail::CompletionState<::morph::wire::ProtocolNegotiationResult>> + negotiation) { + fileControl(::morph::wire::makeHello(), PendingControl{.state = nullptr, + .what = "hello", + .echo = std::nullopt, + .body = nullptr, + .hello = true, + .negotiation = std::move(negotiation)}); } /// The control-call counterpart of `fileExecute`. @@ -672,10 +820,16 @@ class SocketBackend : public ::morph::backend::detail::IBackend { /// Settles every pending call, of both kinds, with @p exc. The tables /// are emptied first, so a settle that re-enters the backend files - /// into empty tables. - void cancelAll(const std::exception_ptr& exc) { + /// into empty tables. With @p onServer, each abandoned execute is + /// also cancelled on the server, when it advertised `"cancel"`. + void cancelAll(const std::exception_ptr& exc, CancelOnServer onServer = CancelOnServer::No) { auto drained = executes.drain(); auto drainedControl = controls.drain(); + if (onServer == CancelOnServer::Yes && cancelAdvertised) { + for (auto const& [callId, pending] : drained) { + sendCancel(callId, pending.stop ? pending.session : session); + } + } for (auto& entry : drained) { if (entry.second.state) { entry.second.state->setException(exc); @@ -694,6 +848,11 @@ class SocketBackend : public ::morph::backend::detail::IBackend { for (auto const& waiter : std::exchange(connectWaiters, {})) { waiter->set_value(); } + if (negotiate) { + // A new connection may reach another server: learn again + // whether it honours `"cancel"`. + fileHello(nullptr); + } if (isReconnect && reconnectHandler && reconnectExec != nullptr) { // Posted, never run here: the handler re-registers through // `bindModel`, whose replies this loop delivers. @@ -703,6 +862,7 @@ class SocketBackend : public ::morph::backend::detail::IBackend { void onDisconnected() { connected.store(false); + cancelAdvertised = false; if (conn) { ::morph::net::detail::closeAfterFlush(conn); conn.reset(); @@ -738,7 +898,13 @@ class SocketBackend : public ::morph::backend::detail::IBackend { connected.store(false); static_cast(loop.loop().cancelTimer(backoffTimer)); if (conn) { - ::morph::net::detail::closeConnection(*conn); + // After the frames already queued are written, not instead of + // them: the owner's `cancelPending` runs just before this + // close, and the cancels and deregisters it queued are what + // tell the server to stop work nobody will read. The read flow + // sees `closed` and stops dispatching meanwhile; the write is + // still bounded by `sendTimeout`. + ::morph::net::detail::closeAfterFlush(conn); conn.reset(); } connectWaiters.clear(); @@ -769,6 +935,9 @@ class SocketBackend : public ::morph::backend::detail::IBackend { // Not an execute: it may be a control reply for a `bindModel`/ // `promoteModel` in flight, which shares this id space. if (auto control = controls.take(env.callId)) { + if (control->hello) { + cancelAdvertised = ::morph::wire::helloAdvertises(env, ::morph::wire::kCapabilityCancel); + } settleControl(*control, env); } return; // otherwise a late/cancelled reply, dropped silently @@ -922,6 +1091,11 @@ class SocketBackend : public ::morph::backend::detail::IBackend { ::morph::exec::IExecutor* reconnectExec{nullptr}; bool everConnected{false}; bool closed{false}; + /// Set by `negotiateProtocolVersion`: re-send `"hello"` on each connect. + bool negotiate{false}; + /// Whether this connection's server advertised `"cancel"`. Cleared on + /// every disconnect, learned again from the next `"hello"` reply. + bool cancelAdvertised{false}; std::chrono::milliseconds reconnectDelay; ::core::net::TimerId backoffTimer{}; std::shared_ptr<::morph::net::detail::LoopConnection> conn; diff --git a/include/morph/qt/qt_websocket_backend.hpp b/include/morph/qt/qt_websocket_backend.hpp index c1d7831e3..a167a2596 100644 --- a/include/morph/qt/qt_websocket_backend.hpp +++ b/include/morph/qt/qt_websocket_backend.hpp @@ -7,8 +7,12 @@ #include #include #include +#include #include +#include +#include #include +#include #include #include #include @@ -173,7 +177,10 @@ class QtWebSocketBackend : public ::morph::backend::detail::IBackend { /// @brief Sends a `"hello"` envelope to the server and classifies its reply. /// /// Intended to be sent once, after the socket connects and before any - /// `bindModel`/`execute`; nothing enforces that ordering. + /// `bindModel`/`execute`; nothing enforces that ordering. The reply also + /// says whether the server honours `"cancel"`, which is the only way this + /// backend ever sends one; once called, `"hello"` is re-sent on every + /// reconnect, since a new connection may reach a different server. /// /// @param replyExec Executor the answer is delivered on. Borrowed: it must /// outlive the returned `Completion`. @@ -414,6 +421,21 @@ class QtWebSocketBackend : public ::morph::backend::detail::IBackend { /// before the reconnect handler fires. void flushQueuedRegistrations(); + /// @brief Sends a `cancel` for every pending execute whose stop was + /// requested since the last drain, if the server advertised + /// `"cancel"`. On the socket's thread, from `_cancelWake`. + void drainCancels(); + + /// @brief Sends a fire-and-forget `cancel` for @p target; its reply is + /// filed nowhere and dropped. + /// @param target Id of the execute to stop. + /// @param callerSession Session the execute carried. + void sendCancel(uint64_t target, const ::morph::session::Context& callerSession); + + /// @brief Re-sends `"hello"` after a reconnect, recording only whether the + /// server advertises `"cancel"`. + void resendHello(); + /// @brief Asserts, in a debug build, that @p site runs on the socket's /// thread — where every table this backend keeps belongs. /// @param site Name of the calling body. @@ -426,6 +448,10 @@ class QtWebSocketBackend : public ::morph::backend::detail::IBackend { Config _cfg; QWebSocket _socket; QTimer _reconnectTimer; + /// Zero-interval single shot whose `timeout` runs `drainCancels` on the + /// socket's thread. A stop callback, on any thread, starts it through a + /// queued `QMetaObject::invokeMethod(&_cancelWake, "start")`. + QTimer _cancelWake; std::chrono::milliseconds _currentReconnectDelay; bool _connected{false}; bool _everConnected{false}; @@ -435,11 +461,36 @@ class QtWebSocketBackend : public ::morph::backend::detail::IBackend { std::function _connectHandler; std::function _disconnectHandler; ::morph::session::Context _session; + /// Set by `negotiateProtocolVersion`: re-send `"hello"` on each connect. + bool _negotiate{false}; + /// Whether this connection's server advertised `"cancel"`. Cleared on + /// every disconnect, learned again from the next `"hello"` reply. + bool _cancelAdvertised{false}; + + /// The stop callback a stoppable execute registers. Runs on whichever + /// thread requested the stop, so it touches no table: it marks its call + /// and queues a start of `_cancelWake`, whose drain runs on the socket's + /// thread. A start queued for a destroyed timer is dropped with it. + struct CancelOnStop { + std::shared_ptr> requested; + QTimer* wake{nullptr}; + void operator()() const noexcept; + }; struct PendingExecute { std::shared_ptr<::morph::async::detail::CompletionState>> state; std::function(std::string_view)> deserialize; ::morph::exec::IExecutor* cbExec{nullptr}; + /// The call's stop source, or null for a call nothing can stop. + std::shared_ptr<::core::async::StopSource> stop; + /// The execute's session, for the cancel; empty without `stop`. + ::morph::session::Context session; + /// Set by the stop callback, cleared by the drain that sends the + /// cancel; null without `stop`. + std::shared_ptr> stopRequested; + /// Registered on `stop`; deregistered when the entry goes, so a stop + /// after the reply sends nothing. + std::shared_ptr<::core::async::StopCallback> watch; }; uint64_t _nextCallId{0}; // Every table below is touched only on the socket's thread: by the verbs, diff --git a/scripts/branch_partial_allowlist.json b/scripts/branch_partial_allowlist.json index 573327f20..19e1fa0f1 100644 --- a/scripts/branch_partial_allowlist.json +++ b/scripts/branch_partial_allowlist.json @@ -115,7 +115,7 @@ }, { "file": "include/morph/net/socket_backend.hpp", - "line": 797, + "line": 966, "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/src/qt/qt_websocket_backend.cpp b/src/qt/qt_websocket_backend.cpp index 25e96413f..73061ed5c 100644 --- a/src/qt/qt_websocket_backend.cpp +++ b/src/qt/qt_websocket_backend.cpp @@ -35,12 +35,20 @@ QtWebSocketBackend::QtWebSocketBackend(QUrl serverUrl, ::morph::model::detail::A #endif _reconnectTimer.setSingleShot(true); QObject::connect(&_reconnectTimer, &QTimer::timeout, [this] { attemptReconnect(); }); + _cancelWake.setSingleShot(true); + _cancelWake.setInterval(0); + QObject::connect(&_cancelWake, &QTimer::timeout, [this] { drainCancels(); }); QObject::connect(&_socket, &QWebSocket::connected, [this]() { const bool isReconnect = _everConnected; _connected = true; _everConnected = true; _currentReconnectDelay = _cfg.initialReconnectDelay; + if (_negotiate) { + // A new connection may reach another server: learn again whether + // it honours "cancel". + resendHello(); + } // Fires on every successful connect, first included -- the general // "transport is up" notification a status indicator wants. if (_connectHandler) { @@ -59,6 +67,7 @@ QtWebSocketBackend::QtWebSocketBackend(QUrl serverUrl, ::morph::model::detail::A }); QObject::connect(&_socket, &QWebSocket::disconnected, [this]() { _connected = false; + _cancelAdvertised = false; // Fires before reconnect scheduling below, so an observer sees the // disconnected state even when a retry follows immediately. if (_disconnectHandler) { @@ -77,8 +86,13 @@ QtWebSocketBackend::QtWebSocketBackend(QUrl serverUrl, ::morph::model::detail::A QtWebSocketBackend::~QtWebSocketBackend() { // NOLINT(modernize-use-equals-default) _shuttingDown = true; _reconnectTimer.stop(); + _cancelWake.stop(); // Disconnect all signals first so no slot tries to access our members after they destruct. _socket.disconnect(); + // Hands the frames already queued -- the cancels the owner's cancelPending + // sent just before -- to the TCP socket before the abort discards them. + // Never blocks: what the kernel will not take now is lost, as before. + _socket.flush(); // Abort cleanly: sends TCP RST without attempting close handshake. _socket.abort(); // Safety net: if the owner did not run cancelPending() first (e.g. backend used @@ -214,10 +228,13 @@ ::morph::async::Completion<::morph::wire::ProtocolNegotiationResult> QtWebSocket promise.reject(std::make_exception_ptr(std::runtime_error{"protocol negotiation failed: disconnected"})); return std::move(completion); } + _negotiate = true; auto shared = std::make_shared<::morph::async::Completion::Promise>(std::move(promise)); sendControl(::morph::wire::makeHello(), PendingControl{.reply = - [shared](const ::morph::wire::Envelope& env) { + [this, shared](const ::morph::wire::Envelope& env) { + _cancelAdvertised = + ::morph::wire::helloAdvertises(env, ::morph::wire::kCapabilityCancel); try { shared->resolve(::morph::wire::interpretHelloReply(env)); } catch (...) { @@ -228,6 +245,46 @@ ::morph::async::Completion<::morph::wire::ProtocolNegotiationResult> QtWebSocket return std::move(completion); } +void QtWebSocketBackend::resendHello() { + sendControl(::morph::wire::makeHello(), PendingControl{.reply = + [this](const ::morph::wire::Envelope& env) { + _cancelAdvertised = ::morph::wire::helloAdvertises( + env, ::morph::wire::kCapabilityCancel); + }, + .fail = [](const std::exception_ptr&) {}}); +} + +void QtWebSocketBackend::CancelOnStop::operator()() const noexcept { + requested->store(true, std::memory_order_release); + // The member-name overload, not a functor: it allocates nothing, and the + // timer's `start` slot runs on the timer's own thread, the socket's. + static_cast(QMetaObject::invokeMethod(wake, "start", Qt::QueuedConnection)); +} + +void QtWebSocketBackend::drainCancels() { + checkThread("QtWebSocketBackend::drainCancels"); + if (_shuttingDown || !_connected || !_cancelAdvertised) { + return; + } + for (auto const& [callId, pending] : _pending) { + if (pending.stopRequested && pending.stopRequested->exchange(false, std::memory_order_acquire)) { + sendCancel(callId, pending.session); + } + } +} + +void QtWebSocketBackend::sendCancel(uint64_t target, const ::morph::session::Context& callerSession) { + auto env = ::morph::wire::makeCancel(target); + env.callId = ++_nextCallId; + env.session = callerSession; + try { + _socket.sendTextMessage(QString::fromStdString(::morph::wire::encode(env))); + } catch (const std::exception&) { // NOLINT(bugprone-empty-catch) + // Best effort, like a deregister: without the cancel the server's + // handler runs to its end, as it did before cancels. + } +} + void QtWebSocketBackend::checkThread(char const* site) const noexcept { // A QObject's state belongs to the thread it lives on. static_cast(site); @@ -330,9 +387,28 @@ ::morph::async::Completion> QtWebSocketBackend::execute( env.session = std::move(call.session); checkThread("QtWebSocketBackend::execute"); - _pending[callId] = PendingExecute{.state = compState, .deserialize = call.deserializeResult, .cbExec = cbExec}; + auto& pending = _pending[callId]; + pending = PendingExecute{.state = compState, .deserialize = call.deserializeResult, .cbExec = cbExec}; + if (call.stopSource) { + // Kept for the cancel a stop sends: the server honours it only under + // the principal the execute was admitted with. + pending.stop = call.stopSource; + pending.session = env.session; + } + auto const stop = pending.stop; _socket.sendTextMessage(QString::fromStdString(::morph::wire::encode(env))); + if (stop) { + // After the frame is sent, so a cancel never precedes its execute. A + // stop already requested runs the callback here; it queues the drain. + auto requested = std::make_shared>(false); + auto watch = std::make_shared<::core::async::StopCallback>( + stop->get_token(), CancelOnStop{.requested = requested, .wake = &_cancelWake}); + if (auto const iter = _pending.find(callId); iter != _pending.end()) { + iter->second.stopRequested = std::move(requested); + iter->second.watch = std::move(watch); + } + } return comp; } @@ -343,6 +419,13 @@ void QtWebSocketBackend::cancelPending(const std::exception_ptr& exc) { auto drainedExecutes = std::exchange(_pending, {}); auto drainedControl = std::exchange(_pendingControl, {}); auto drainedQueue = std::exchange(_queuedRegistrations, {}); + // Each abandoned execute is cancelled on the server while the socket is + // up. Not from the destructor: its socket is already aborted. + if (_connected && _cancelAdvertised && !_shuttingDown) { + for (auto const& [callId, pending] : drainedExecutes) { + sendCancel(callId, pending.stop ? pending.session : _session); + } + } // _pendingDeregisters tracks fire-and-forget requests nobody awaits -- // just drop the bookkeeping, there is no callback to invoke. _pendingDeregisters.clear(); diff --git a/tests/net/test_socket_backend.cpp b/tests/net/test_socket_backend.cpp index a180b57aa..2e3773c38 100644 --- a/tests/net/test_socket_backend.cpp +++ b/tests/net/test_socket_backend.cpp @@ -8,6 +8,11 @@ #include #include #include +#include +#include +#include +#include +#include #include #include #include @@ -29,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -2430,3 +2436,203 @@ TEST_CASE("SocketBackend: a synchronous verb called on the loop's own thread fai }); CHECK_THAT(message, Catch::Matchers::ContainsSubstring("cannot wait on the I/O loop's own thread")); } + +// ── Cancelling a call on the server ───────────────────────────────────────── +// +// A client that gives up on a call -- its deadline passed, its bridge went +// away -- sends `cancel` to a server that advertised it, so the server's Task +// handler is stopped rather than left running for a reply nobody reads. + +struct ScPark { + int tag = 0; +}; + +/// What the server-side handler reports back, read by the test thread. +struct ScParkProbe { + std::atomic started{0}; + std::atomic stopped{0}; + std::atomic finished{0}; + std::unique_ptr> queue; +}; + +inline ScParkProbe& scParkProbe() { + static ScParkProbe instance; + return instance; +} + +/// A Task handler that blocks until an item arrives, or until it is stopped. +struct ScParkModel { + // NOLINTNEXTLINE(readability-convert-member-functions-to-static) + core::async::Task execute(ScPark action) { + scParkProbe().started.fetch_add(1); + try { + auto const item = co_await scParkProbe().queue->pop(); + scParkProbe().finished.fetch_add(1); + co_return item.value_or(0) + action.tag; + } catch (const core::async::OperationCancelled&) { + scParkProbe().stopped.fetch_add(1); + scParkProbe().finished.fetch_add(1); + throw; + } + } +}; + +BRIDGE_REGISTER_MODEL(ScParkModel, "ScParkModel") +BRIDGE_REGISTER_ACTION(ScParkModel, ScPark, "ScPark") + +namespace { + +/// A real SocketServer over a RemoteServer, and the queue the parking handler +/// waits on. Unparks a handler a failed case leaves behind before the queue goes. +struct ParkServer { + core::async::ThreadPoolExecutor foreign{1}; + morph::exec::ThreadPoolExecutor serverPool{2}; + std::shared_ptr server = std::make_shared(serverPool); + morph::net::SocketServer wsServer{*server, 0}; + + ParkServer() { + scParkProbe().started = 0; + scParkProbe().stopped = 0; + scParkProbe().finished = 0; + scParkProbe().queue = + std::make_unique>(foreign, core::async::AsyncQueueOptions{}); + REQUIRE(wsServer.listen()); + } + ParkServer(const ParkServer&) = delete; + ParkServer& operator=(const ParkServer&) = delete; + ParkServer(ParkServer&&) = delete; + ParkServer& operator=(ParkServer&&) = delete; + // NOLINTNEXTLINE(bugprone-exception-escape): a teardown failure here has no recovery; terminating is the outcome. + ~ParkServer() { + scParkProbe().queue->close(); + static_cast( + morph::testing::waitUntil([] { return scParkProbe().finished.load() == scParkProbe().started.load(); })); + wsServer.close(); + scParkProbe().queue.reset(); + } + + [[nodiscard]] std::string url() const { + return "ws://127.0.0.1:" + std::to_string(static_cast(wsServer.port())); + } +}; + +/// Negotiates on @p backend, pumping @p owner for the answer. +morph::wire::ProtocolNegotiationResult negotiate(morph::net::SocketBackend& backend, + morph::exec::MainThreadExecutor& owner) { + std::optional result; + auto completion = backend.negotiateProtocolVersion(owner); + std::move(completion) + .then([&](morph::wire::ProtocolNegotiationResult value) { result = value; }) + .onError([](const std::exception_ptr&) {}); + spinUntil(owner, [&] { return result.has_value(); }); + REQUIRE(result.has_value()); + return *result; +} + +} // namespace + +TEST_CASE("SocketBackend: a client deadline stops the server's Task handler", "[net][socket_backend][cancel]") { + ParkServer const park; + auto backendPtr = std::make_unique(park.url()); + REQUIRE(backendPtr->waitForConnected()); + morph::exec::MainThreadExecutor owner; + REQUIRE(negotiate(*backendPtr, owner) == morph::wire::ProtocolNegotiationResult::Negotiated); + morph::bridge::Bridge bridge{std::move(backendPtr), owner}; + bridge.setExecuteDeadline(std::chrono::milliseconds{100}); + morph::bridge::BridgeHandler handler{bridge, &owner}; + + std::atomic timedOut{false}; + handler.execute(ScPark{}).then([](int) {}).onError([&](const std::exception_ptr& error) { + try { + std::rethrow_exception(error); + } catch (const morph::backend::ClientTimeoutError&) { + timedOut = true; + } catch (...) { // NOLINT(bugprone-empty-catch): any other failure leaves timedOut false + } + }); + + // Nothing but the client's cancel can end the server's handler: its queue + // is never pushed to. + spinUntil(owner, [&] { return timedOut.load() && scParkProbe().stopped.load() == 1; }); + CHECK(timedOut.load()); + CHECK(scParkProbe().stopped.load() == 1); +} + +TEST_CASE("SocketBackend: destroying the bridge cancels its calls on the server", "[net][socket_backend][cancel]") { + ParkServer const park; + auto backendPtr = std::make_unique(park.url()); + REQUIRE(backendPtr->waitForConnected()); + morph::exec::MainThreadExecutor owner; + REQUIRE(negotiate(*backendPtr, owner) == morph::wire::ProtocolNegotiationResult::Negotiated); + { + morph::bridge::Bridge bridge{std::move(backendPtr), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + handler.execute(ScPark{}).then([](int) {}).onError([](const std::exception_ptr&) {}); + spinUntil(owner, [] { return scParkProbe().started.load() == 1; }); + REQUIRE(scParkProbe().started.load() == 1); + } + // ~Bridge calls cancelPending, which sends the cancel from the loop before + // the backend's own close. + spinUntil([] { return scParkProbe().stopped.load() == 1; }); + CHECK(scParkProbe().stopped.load() == 1); +} + +TEST_CASE("SocketBackend: without a negotiated cancel, a deadline leaves the server's handler running", + "[net][socket_backend][cancel]") { + ParkServer const park; + auto backendPtr = std::make_unique(park.url()); + REQUIRE(backendPtr->waitForConnected()); + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::move(backendPtr), owner}; + bridge.setExecuteDeadline(std::chrono::milliseconds{50}); + morph::bridge::BridgeHandler handler{bridge, &owner}; + + std::atomic failed{false}; + handler.execute(ScPark{}).then([](int) {}).onError([&](const std::exception_ptr&) { failed = true; }); + spinUntil(owner, [&] { return failed.load(); }); + REQUIRE(failed.load()); + // The deadline has fired client-side; the server's handler is still + // parked, and ends with the item rather than a stop. + spinUntil(owner, [] { return scParkProbe().started.load() == 1; }); + static_cast(scParkProbe().queue->push(41)); + spinUntil([] { return scParkProbe().finished.load() == 1; }); + CHECK(scParkProbe().finished.load() == 1); + CHECK(scParkProbe().stopped.load() == 0); +} + +TEST_CASE("SocketBackend: never sends cancel to a server that did not advertise it", "[net][socket_backend][cancel]") { + FakeWsServer fake; + morph::net::SocketBackend backend{"ws://127.0.0.1:" + std::to_string(static_cast(fake.port()))}; + fake.acceptAndHandshake(); + REQUIRE(backend.waitForConnected()); + morph::exec::MainThreadExecutor owner; + + // A legacy server: it does not know "hello". + std::optional result; + backend.negotiateProtocolVersion(owner) + .then([&](morph::wire::ProtocolNegotiationResult value) { result = value; }) + .onError([](const std::exception_ptr&) {}); + auto const hello = fake.receiveEnvelope(); + REQUIRE(hello.kind == "hello"); + fake.sendFrame(morph::net::detail::WsOpcode::kText, + morph::wire::encode(morph::wire::makeErr("unknown envelope kind: hello", hello.callId))); + spinUntil(owner, [&] { return result.has_value(); }); + REQUIRE(result == morph::wire::ProtocolNegotiationResult::LegacyPeer); + + // A stoppable execute, stopped, then swept by cancelPending: either would + // send a cancel to a server that advertised one. + auto const stop = std::make_shared(); + morph::backend::detail::ActionCall call; + call.modelTypeId = "ScParkModel"; + call.actionTypeId = "ScPark"; + call.action = std::make_shared(0); + call.serializeAction = [](const void*) { return std::string{"{}"}; }; + call.stopSource = stop; + auto const completion = backend.execute(morph::exec::detail::ModelId{7}, std::move(call), &owner); + CHECK(fake.receiveEnvelope().kind == "execute"); + static_cast(stop->request_stop()); + backend.cancelPending(std::make_exception_ptr(std::runtime_error{"cancelled"})); + // A marker the backend sends after both: the next frame the server reads. + backend.deregisterModel(morph::exec::detail::ModelId{7}); + CHECK(fake.receiveEnvelope().kind == "deregister"); +} diff --git a/tests/qt/test_qt_websocket.cpp b/tests/qt/test_qt_websocket.cpp index 7617a1eb7..7a799f87f 100644 --- a/tests/qt/test_qt_websocket.cpp +++ b/tests/qt/test_qt_websocket.cpp @@ -18,6 +18,10 @@ #include #include #include +#include +#include +#include +#include #include #include #include @@ -2755,3 +2759,120 @@ TEST_CASE("morph::qt::QtWebSocketBackend: the non-blocking control envelopes car // session and could not authenticate the caller at all. CHECK(authorizer->registerTokens.back() == "tok-495"); } + +// ── Cancelling a call on the server ───────────────────────────────────────── +// +// A client that gives up on a call sends `cancel` to a server that advertised +// it, so the server's Task handler is stopped rather than left running. + +struct WsPark { + int tag = 0; +}; + +/// What the server-side handler reports back, read by the test thread. +struct WsParkProbe { + std::atomic started{0}; + std::atomic stopped{0}; + std::atomic finished{0}; + std::unique_ptr> queue; +}; + +inline WsParkProbe& wsParkProbe() { + static WsParkProbe instance; + return instance; +} + +/// A Task handler that blocks until an item arrives, or until it is stopped. +struct WsParkModel { + // NOLINTNEXTLINE(readability-convert-member-functions-to-static) + core::async::Task execute(WsPark action) { + wsParkProbe().started.fetch_add(1); + try { + auto const item = co_await wsParkProbe().queue->pop(); + wsParkProbe().finished.fetch_add(1); + co_return item.value_or(0) + action.tag; + } catch (const core::async::OperationCancelled&) { + wsParkProbe().stopped.fetch_add(1); + wsParkProbe().finished.fetch_add(1); + throw; + } + } +}; + +BRIDGE_REGISTER_MODEL(WsParkModel, "WsParkModel") +BRIDGE_REGISTER_ACTION(WsParkModel, WsPark, "WsPark") + +namespace { + +/// A real QtWebSocketServer over a RemoteServer, and the queue the parking +/// handler waits on. Unparks a handler a failed case leaves behind. +struct WsParkServer { + core::async::ThreadPoolExecutor foreign{1}; + morph::exec::ThreadPoolExecutor serverPool{2}; + std::shared_ptr server = std::make_shared(serverPool); + morph::qt::QtWebSocketServer wsServer{*server, 0}; + + WsParkServer() { + wsParkProbe().started = 0; + wsParkProbe().stopped = 0; + wsParkProbe().finished = 0; + wsParkProbe().queue = + std::make_unique>(foreign, core::async::AsyncQueueOptions{}); + REQUIRE(wsServer.listen()); + } + WsParkServer(const WsParkServer&) = delete; + WsParkServer& operator=(const WsParkServer&) = delete; + WsParkServer(WsParkServer&&) = delete; + WsParkServer& operator=(WsParkServer&&) = delete; + ~WsParkServer() { + wsParkProbe().queue->close(); + pumpUntil([] { return wsParkProbe().finished.load() == wsParkProbe().started.load(); }, 300); + wsParkProbe().queue.reset(); + } + + [[nodiscard]] QUrl url() const { return QUrl{QString("ws://127.0.0.1:%1").arg(wsServer.port())}; } +}; + +} // namespace + +TEST_CASE("morph::qt::QtWebSocketBackend: a client deadline stops the server's Task handler", "[qt][ws][cancel]") { + ensureApp(); + WsParkServer const park; + auto backendPtr = std::make_unique(park.url()); + REQUIRE(backendPtr->waitForConnected()); + morph::exec::MainThreadExecutor cbExec; + REQUIRE(awaitOn(cbExec, backendPtr->negotiateProtocolVersion(cbExec)) == + morph::wire::ProtocolNegotiationResult::Negotiated); + + morph::qt::QtExecutor qtExec; + morph::bridge::Bridge bridge{std::move(backendPtr), qtExec}; + bridge.setExecuteDeadline(std::chrono::milliseconds{100}); + std::atomic failed{false}; + morph::bridge::BridgeHandler handler{bridge, &qtExec}; + handler.execute(WsPark{}).then([](int) {}).onError([&](const std::exception_ptr&) { failed = true; }); + + // Nothing but the client's cancel can end the server's handler. + pumpUntil([&] { return failed.load() && wsParkProbe().stopped.load() == 1; }, 300); + CHECK(failed.load()); + CHECK(wsParkProbe().stopped.load() == 1); +} + +TEST_CASE("morph::qt::QtWebSocketBackend: destroying the bridge cancels its calls on the server", "[qt][ws][cancel]") { + ensureApp(); + WsParkServer const park; + auto backendPtr = std::make_unique(park.url()); + REQUIRE(backendPtr->waitForConnected()); + morph::exec::MainThreadExecutor cbExec; + REQUIRE(awaitOn(cbExec, backendPtr->negotiateProtocolVersion(cbExec)) == + morph::wire::ProtocolNegotiationResult::Negotiated); + { + morph::qt::QtExecutor qtExec; + morph::bridge::Bridge bridge{std::move(backendPtr), qtExec}; + morph::bridge::BridgeHandler handler{bridge, &qtExec}; + handler.execute(WsPark{}).then([](int) {}).onError([](const std::exception_ptr&) {}); + pumpUntil([] { return wsParkProbe().started.load() == 1; }, 300); + REQUIRE(wsParkProbe().started.load() == 1); + } + pumpUntil([] { return wsParkProbe().stopped.load() == 1; }, 300); + CHECK(wsParkProbe().stopped.load() == 1); +}