diff --git a/docs/spec/core/backend.md b/docs/spec/core/backend.md index cd0eadd0f..1641cd04d 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 @@ -1214,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 @@ -1554,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 @@ -1857,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. | @@ -1913,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`. | @@ -1922,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 0027ab667..d772bf4a1 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,112 @@ 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. + +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. + +### 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 +790,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 +804,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 +812,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/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/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/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); +} 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