From 4c77fde77cfcb06cb87f0d519ab1c8feadb67f55 Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Sun, 4 Oct 2026 01:46:37 +0200 Subject: [PATCH 1/3] core: a stopped co_await stops the call it was waiting on (#862) CompletionAwaiter now links its scope to the awaited call's stop source through CompletionState::linkStop when it attaches, as a scope-gated continuation already does. A stop that withdraws the await, or the frame destroyed while suspended, therefore also asks the call to stop: a Task handler on a LocalBackend sees OperationCancelled at its next stop-aware await. linkStop alone was not the whole fix. A stop that lands before the handlers are attached (a token already stopped at co_await, or a stop racing the stop-callback registration) never reaches attach(), so the awaiter now relays that stop to the call on the completion's executor. The cancellation policy gains a measured G2 row for a stopped co_await; the coroutine spec's Cancellation section says what the stop now does to the call. Co-Authored-By: Claude Opus 5.5 --- docs/spec/concurrency_and_lifetimes.md | 11 +- docs/spec/core/coroutines.md | 18 ++- .../morph/core/detail/completion_awaiter.hpp | 29 +++- tests/test_cancellation_policy.cpp | 145 +++++++++++++++++- tests/test_coroutine_client.cpp | 23 +++ 5 files changed, 210 insertions(+), 16 deletions(-) diff --git a/docs/spec/concurrency_and_lifetimes.md b/docs/spec/concurrency_and_lifetimes.md index 050337fe9..5ea2de85a 100644 --- a/docs/spec/concurrency_and_lifetimes.md +++ b/docs/spec/concurrency_and_lifetimes.md @@ -410,6 +410,7 @@ the "work" column below is for a synchronous handler unless it says otherwise. | `Bridge::setExecuteDeadline` | **G0** for calls already made | Applies to calls made after it. When a call's deadline fires, the call ends as in G2, with `ClientTimeoutError` | A synchronous action runs to its end; a Task handler is asked to stop | M | | `BridgeHandler::unsubscribe()` | **G1** | A delivery already posted still runs after it returns; no new one is scheduled. Pair it with a `CallbackScope` for G3 | n/a | M | | `CallbackScope::requestStop()`, `reset()`, `~CallbackScope` | **G3** on the delivery executor; **G1** from another thread | A gated callback posted but not yet run is refused when it runs. From another thread the check and the body are not atomic, so a callback that has passed its check still runs — advisory, and not measured as a race | Never waits for a callback in its body (measured). Stops every call a gated continuation is attached to | M | +| A stop on a coroutine suspended in `co_await completion` | **G2** for the await | The coroutine's resumption with `OperationCancelled` is posted to the executor it suspended on; the await's handlers are refused, so the call's own outcome, later, does not reach it | The call is asked to stop, through the awaiter's scope linked to its stop source: a Task handler on a `LocalBackend` sees the stop (measured). Not waited for. A synchronous action runs to its end; a remote server's handler runs on | M | | `TimeoutScheduler::cancel` | **G0** on return; **G3** for a callback not yet started, once the loop has run the cancel | The cancel is posted to the loop: on return the callback still holds its captures | Never waits for a callback already running | M | | `~TimeoutScheduler` | **G4** | Every pending callback is dropped unfired | Returns only once no callback of the scheduler is running | M | | `LimitPolicy::executeTimeout` (`RemoteServer`) | **G2** for the reply | `err "timeout"` is sent; the handler's own reply, later, is dropped — one reply per call | A synchronous handler runs to its end; a Task handler is asked to stop | M | @@ -425,14 +426,10 @@ the "work" column below is for a synchronous handler unless it says otherwise. - **No cancel crosses the wire.** A remote call is settled locally only; the server finishes the handler and its reply is dropped. There is no `cancel` envelope. -- **A `co_await` that is stopped withdraws the await, not the call.** The - coroutine resumes with `OperationCancelled`; the call it was waiting on runs - on, because the awaiter's stop is not linked to the call's stop source. A - continuation attached through a `CallbackScope` *is* linked: stopping the - scope stops the call. - **A caller has no direct cancel verb on one call.** `Completion` has no - cancel API and `execute()` takes no stop token; the scope link above and the - execute deadline are the only per-call routes. + cancel API and `execute()` takes no stop token; a stopped scope a + continuation is gated by, a stopped `co_await`, and the execute deadline are + the only per-call routes. ## Gating callbacks on a receiver's lifetime — `CallbackScope` diff --git a/docs/spec/core/coroutines.md b/docs/spec/core/coroutines.md index 26dc315db..4e06aa75c 100644 --- a/docs/spec/core/coroutines.md +++ b/docs/spec/core/coroutines.md @@ -167,10 +167,20 @@ a stop callback for the duration of the suspension: executor it suspended on, or on the completion's executor. - **A token already stopped at `co_await`** resumes with `OperationCancelled` without attaching anything to the completion. - -The completion itself is not cancelled: it is still settled by whatever -produces it, and its other handlers still run. Cancellation withdraws this one -await. +- **A stop also stops the call.** The await owns the call it waits on, the way + a continuation attached through a `CallbackScope` does: the awaiter's scope + is linked to the call's stop source when the handlers are attached + (`CompletionState::linkStop`), so the stop that withdraws the await — or the + frame destroyed while suspended — asks the call to stop too. A stop that + lands before the handlers are attached, a token already stopped included, + is relayed to the call on the completion's executor. Only a call that + carries a stop source can be stopped: a `Bridge` call whose handler returns + a `Task`, on a `LocalBackend`. Its handler sees `OperationCancelled` at its + next stop-aware `co_await`. + +The completion itself is still settled by whatever produces it — a stopped +Task handler rejects it with `OperationCancelled` — and its other handlers +still run; the withdrawn await's own handlers do not. ### `spawn` — the entry point from non-coroutine code diff --git a/include/morph/core/detail/completion_awaiter.hpp b/include/morph/core/detail/completion_awaiter.hpp index 17b9d0940..f0fefd226 100644 --- a/include/morph/core/detail/completion_awaiter.hpp +++ b/include/morph/core/detail/completion_awaiter.hpp @@ -37,7 +37,8 @@ struct CompletionState; /// `CallbackToken`. With a stoppable token on the awaiting promise it also /// registers a stop callback. Exactly one of *the completion's handler* and /// *the stop callback* moves `Shared::outcome` off `Pending`, and only that one -/// resumes the coroutine. +/// resumes the coroutine. A stop that withdraws the await also stops the call +/// the completion reports on, when that call carries a stop source. /// /// The pair is attached on the completion's executor, which owns its handler /// lists: directly when the awaiting coroutine is running there, otherwise by @@ -169,6 +170,7 @@ class CompletionAwaiter { ::core::async::StopToken const token = awaiting.promise().stopToken(); if (token.stop_requested()) { shared->outcome.store(Outcome::Cancelled); + stopUnattached(state, token); return false; } if (token.stop_possible()) { @@ -177,6 +179,7 @@ class CompletionAwaiter { if (shared->outcome.load(std::memory_order_seq_cst) == Outcome::Cancelled) { // A stop landed while registering. Whoever claims answers // it; if onStop claimed first it is resuming us already. + stopUnattached(state, token); return shared->claimed.exchange(true, std::memory_order_acq_rel); } } @@ -241,8 +244,15 @@ class CompletionAwaiter { private: /// Attaches the await's handler pair to @p state; runs on its executor. + /// + /// The await owns the call it waits on, as a scope-gated continuation + /// does: the scope's stop -- a stop that withdraws the await, or the frame + /// destroyed while suspended -- is linked to the call's stop source, so + /// withdrawing the await also asks the call to stop. Linked after the + /// scope has stopped, it stops the call at once. static void attach(CompletionState& state, const std::shared_ptr& shared) { auto const token = shared->scope.token(); + state.linkStop(token.stopToken()); state.attachThen([token, shared](const T& settled) { if (token.active() && shared->decide(Outcome::Settled)) { shared->value.emplace(settled); @@ -257,6 +267,23 @@ class CompletionAwaiter { }); } + /// Stops the call @p state reports on, for an await a stop withdrew before + /// it attached, so no link carries that stop to the call. @p stopped, + /// already stopped, is linked on the completion's executor, which stops the + /// call there. Best effort: an executor that refuses the task leaves the + /// call running; the coroutine resumes with the cancellation either way. + static void stopUnattached(const std::shared_ptr>& state, + const ::core::async::StopToken& stopped) noexcept { + try { + if (::morph::exec::runningOn(*state->cbExec)) { + state->linkStop(stopped); + return; + } + state->cbExec->post([state, stopped] { state->linkStop(stopped); }); + } catch (...) { // NOLINT(bugprone-empty-catch): the await is answered already + } + } + std::shared_ptr> _state; std::shared_ptr _shared; }; diff --git a/tests/test_cancellation_policy.cpp b/tests/test_cancellation_policy.cpp index 072984589..42373bb2d 100644 --- a/tests/test_cancellation_policy.cpp +++ b/tests/test_cancellation_policy.cpp @@ -12,15 +12,20 @@ // G3 no further callback will start // G4 G3, plus none is still running // -// "Work" is the model action or server handler the verb abandons. Whether a -// Task handler is asked to stop is measured in test_coroutine_model.cpp; these -// cases use a synchronous handler, which nothing can stop, so they measure -// what each verb does to the callbacks and whether it waits for that work. +// "Work" is the model action or server handler the verb abandons. Most cases +// use a synchronous handler, which nothing can stop, so they measure what each +// verb does to the callbacks and whether it waits for that work. The verbs +// whose point is to stop the work -- a stopped co_await -- are measured +// against a Task handler, which can observe a stop; the other verbs' Task +// handler stops are measured in test_coroutine_model.cpp. #include #include #include #include +#include +#include +#include #include #include #include @@ -30,6 +35,7 @@ #include #include #include +#include #include #include #include @@ -60,6 +66,10 @@ struct CPBlock { struct CPFast { int tag = 0; }; +/// Handled by a Task handler that sleeps until it is stopped. +struct CPSleep { + int ms = 0; +}; namespace { @@ -82,6 +92,39 @@ CPGate& gate() { return instance; } +/// The Task handler's progress, and the scheduler it sleeps on. +struct CPSleeper { + std::atomic started{0}; + std::atomic cancelled{0}; + std::atomic finished{0}; + morph::async::detail::TimeoutScheduler* scheduler = nullptr; +}; + +CPSleeper& sleeper() { + static CPSleeper instance; + return instance; +} + +/// Owns the scheduler `CPSleep` waits on for one test, and joins its thread +/// when the test ends rather than keeping it for the rest of the run. +class SleeperScope { +public: + SleeperScope() { + sleeper().started = 0; + sleeper().cancelled = 0; + sleeper().finished = 0; + sleeper().scheduler = &_scheduler; + } + SleeperScope(const SleeperScope&) = delete; + SleeperScope& operator=(const SleeperScope&) = delete; + SleeperScope(SleeperScope&&) = delete; + SleeperScope& operator=(SleeperScope&&) = delete; + ~SleeperScope() { sleeper().scheduler = nullptr; } + +private: + morph::async::detail::TimeoutScheduler _scheduler; +}; + template [[nodiscard]] bool holds(const std::exception_ptr& error) { try { @@ -116,12 +159,24 @@ struct CPModel { return action.tag; } int execute(CPFast action) { return action.tag; } + core::async::Task execute(CPSleep action) { + sleeper().started.fetch_add(1); + try { + co_await morph::async::delay(*sleeper().scheduler, std::chrono::milliseconds{action.ms}); + } catch (const core::async::OperationCancelled&) { + sleeper().cancelled.fetch_add(1); + throw; + } + sleeper().finished.fetch_add(1); + co_return action.ms; + } // NOLINTEND(readability-convert-member-functions-to-static) }; BRIDGE_REGISTER_MODEL(CPModel, "CP_Model") BRIDGE_REGISTER_ACTION(CPModel, CPBlock, "CP_Block") BRIDGE_REGISTER_ACTION(CPModel, CPFast, "CP_Fast") +BRIDGE_REGISTER_ACTION(CPModel, CPSleep, "CP_Sleep") namespace { @@ -478,6 +533,88 @@ TEST_CASE("CallbackScope stop is G3 on the delivery executor, and never waits fo } } +// ── A stopped co_await ────────────────────────────────────────────────────── + +namespace { + +/// What reached a coroutine awaiting one call, on the owner. +struct AwaitSeen { + /// Set in the step that suspends, just before the `co_await`. + std::atomic suspending{false}; + std::atomic finished{false}; + bool cancelled = false; + /// Anything other than the cancellation: a value, or another error. + bool settled = false; +}; + +core::async::Task awaitCall(morph::async::Completion call, std::shared_ptr seen) { + try { + seen->suspending = true; + static_cast(co_await std::move(call)); + seen->settled = true; + } catch (const core::async::OperationCancelled&) { + seen->cancelled = true; + } catch (...) { + seen->settled = true; + } + seen->finished = true; +} + +} // namespace + +TEST_CASE("A stopped co_await is G2 for the await, and stops the Task handler it was waiting on", "[cancel-policy]") { + SleeperScope const sleeping; + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + auto seen = std::make_shared(); + auto task = awaitCall(handler.execute(CPSleep{.ms = 60'000}), seen); + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + task.handle().promise().setStopToken(stop.get_token()); + morph::async::spawn(owner, std::move(task)); + REQUIRE(pumpUntil(owner, [&] { return seen->suspending.load() && sleeper().started.load() == 1; })); + + static_cast(stop.request_stop()); + + // Not G3: the coroutine has not resumed on return. + CHECK_FALSE(seen->finished.load()); + // G2: its resumption with OperationCancelled is posted, and runs on a pump. + REQUIRE(pumpUntil(owner, [&] { return seen->finished.load(); })); + CHECK(seen->cancelled); + // The work: the handler the await was waiting on is asked to stop. + REQUIRE(pumpUntil(owner, [] { return sleeper().cancelled.load() == 1; })); + CHECK(sleeper().finished.load() == 0); + // G1: the call's own outcome, settled by the stopped handler, reaches + // nothing. + owner.runFor(30ms); + CHECK_FALSE(seen->settled); +} + +TEST_CASE("A co_await under a stop already requested stops the Task handler it would have waited on", + "[cancel-policy]") { + SleeperScope const sleeping; + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + auto seen = std::make_shared(); + auto task = awaitCall(handler.execute(CPSleep{.ms = 60'000}), seen); + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + static_cast(stop.request_stop()); + task.handle().promise().setStopToken(stop.get_token()); + morph::async::spawn(owner, std::move(task)); + + REQUIRE(pumpUntil(owner, [&] { return seen->finished.load(); })); + CHECK(seen->cancelled); + REQUIRE(pumpUntil(owner, [] { return sleeper().cancelled.load() == 1; })); + CHECK(sleeper().finished.load() == 0); +} + // ── TimeoutScheduler ──────────────────────────────────────────────────────── TEST_CASE("TimeoutScheduler::cancel is G0 on return and G3 once the loop has run it; it never waits", diff --git a/tests/test_coroutine_client.cpp b/tests/test_coroutine_client.cpp index f412fd516..d31695242 100644 --- a/tests/test_coroutine_client.cpp +++ b/tests/test_coroutine_client.cpp @@ -191,6 +191,29 @@ TEST_CASE("co_await under a stop already requested does not suspend", "[coroutin REQUIRE_FALSE(seen->value.has_value()); } +TEST_CASE("a stop already requested at co_await off the completion's executor stops the call there", + "[coroutine][client]") { + morph::exec::MainThreadExecutor exec; + auto const state = std::make_shared>(); + state->stopSource = std::make_shared(); + auto const seen = std::make_shared(); + + auto const task = awaitInto(Completion{state, &exec}, seen); + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + static_cast(stop.request_stop()); + task.handle().promise().setStopToken(stop.get_token()); + // Resumed by hand, outside every executor's task: the await answers at + // once, and the stop is relayed to the call through the completion's + // executor, not here. + task.handle().resume(); + REQUIRE(seen->finished.load()); + REQUIRE(seen->cancelled); + CHECK_FALSE(state->stopSource->stop_requested()); + + REQUIRE(pumpUntil(exec, [&] { return state->stopSource->stop_requested(); })); +} + TEST_CASE("a rejection after a stop withdrew the await reaches nothing", "[coroutine][client]") { morph::exec::MainThreadExecutor exec; auto [completion, promise] = Completion::makeSettleable(&exec); From ad79b11540568b057b7be65bbf5df4f638b356fb Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Sun, 4 Oct 2026 01:48:33 +0200 Subject: [PATCH 2/3] core: a caller-facing cancel, BridgeHandler::execute(action, StopToken) (#863) A stop on the token before the call settles rejects the Completion with core::async::OperationCancelled, posted from the thread that requested the stop, and requests stop on the call's StopSource, so a Task handler on a LocalBackend sees it. A synchronous handler runs to its end; on a remote backend the call is only abandoned, since no cancel crosses the wire yet. A token already stopped rejects the call without dispatching. The link is CompletionState::linkCancel, held with the scope links in stopLinks and released by deliver(). Like the execute deadline it settles the state, not the sink, so the call stays counted pending until the backend's own reply. The cancellation policy gains a measured G2 row for the verb, stating the remote limitation; completion.md and bridge.md no longer say that no work cancellation exists. Co-Authored-By: Claude Opus 5.5 --- docs/spec/concurrency_and_lifetimes.md | 8 +- docs/spec/core/bridge.md | 3 +- docs/spec/core/completion.md | 31 +++-- include/morph/core/bridge.hpp | 38 ++++++ include/morph/core/completion.hpp | 39 ++++++ scripts/branch_partial_allowlist.json | 2 +- tests/test_cancellation_policy.cpp | 176 ++++++++++++++++++++++++- 7 files changed, 276 insertions(+), 21 deletions(-) diff --git a/docs/spec/concurrency_and_lifetimes.md b/docs/spec/concurrency_and_lifetimes.md index 5ea2de85a..0ce457e23 100644 --- a/docs/spec/concurrency_and_lifetimes.md +++ b/docs/spec/concurrency_and_lifetimes.md @@ -408,6 +408,7 @@ the "work" column below is for a synchronous handler unless it says otherwise. | `~Bridge` | **G2** for its calls | Calls waiting for a bind and the backend's pending calls end with `BridgeDestroyedError`, posted | **G4** with a `LocalBackend`: the backend is destroyed with the bridge, and drains its strands, so the destructor returns only once the running action has ended. A Task handler is asked to stop first | M | | `Bridge::switchBackend` | **G2** for the outgoing backend's calls | `BackendChangedError`, posted | The outgoing backend is destroyed inside the switch: with a `LocalBackend`, the switch waits for its running actions | M | | `Bridge::setExecuteDeadline` | **G0** for calls already made | Applies to calls made after it. When a call's deadline fires, the call ends as in G2, with `ClientTimeoutError` | A synchronous action runs to its end; a Task handler is asked to stop | M | +| A stop on `BridgeHandler::execute(action, stop)`'s token | **G2**, from any thread | `OperationCancelled` is posted to the call's `cbExec` by the thread that requested the stop; the call's own outcome, later, reaches nobody. A token already stopped rejects the call the same way without dispatching it | Asked to stop, not waited for: a Task handler on a `LocalBackend` sees the stop. A synchronous action runs to its end. **On a remote backend the call is only abandoned**: no `cancel` crosses the wire, so the server's handler runs to its end and its reply is dropped (measured with `SimulatedRemoteBackend`). The call stays counted as pending until that reply, as for a fired deadline | M | | `BridgeHandler::unsubscribe()` | **G1** | A delivery already posted still runs after it returns; no new one is scheduled. Pair it with a `CallbackScope` for G3 | n/a | M | | `CallbackScope::requestStop()`, `reset()`, `~CallbackScope` | **G3** on the delivery executor; **G1** from another thread | A gated callback posted but not yet run is refused when it runs. From another thread the check and the body are not atomic, so a callback that has passed its check still runs — advisory, and not measured as a race | Never waits for a callback in its body (measured). Stops every call a gated continuation is attached to | M | | A stop on a coroutine suspended in `co_await completion` | **G2** for the await | The coroutine's resumption with `OperationCancelled` is posted to the executor it suspended on; the await's handlers are refused, so the call's own outcome, later, does not reach it | The call is asked to stop, through the awaiter's scope linked to its stop source: a Task handler on a `LocalBackend` sees the stop (measured). Not waited for. A synchronous action runs to its end; a remote server's handler runs on | M | @@ -426,10 +427,9 @@ the "work" column below is for a synchronous handler unless it says otherwise. - **No cancel crosses the wire.** A remote call is settled locally only; the server finishes the handler and its reply is dropped. There is no `cancel` envelope. -- **A caller has no direct cancel verb on one call.** `Completion` has no - cancel API and `execute()` takes no stop token; a stopped scope a - continuation is gated by, a stopped `co_await`, and the execute deadline are - the only per-call routes. +- **`Completion` has no cancel API of its own.** A call is cancelled through + the token given to `execute(action, stop)`, a stopped scope a continuation is + gated by, a stopped `co_await`, or the execute deadline. ## Gating callbacks on a receiver's lifetime — `CallbackScope` diff --git a/docs/spec/core/bridge.md b/docs/spec/core/bridge.md index 0d086d7f3..793a490ee 100644 --- a/docs/spec/core/bridge.md +++ b/docs/spec/core/bridge.md @@ -868,7 +868,7 @@ is tolerated, as above. | `registerHandler(binding)` | `void registerHandler(const shared_ptr&)` | Pre-built binding. Same bind rule. | | `switchBackend` | `void switchBackend(unique_ptr)` / `void switchBackend(shared_ptr)` | Tells the new backend its owner and pushes the default session, then issues a re-bind for every live binding. Binds settled before returning decide it: a failure releases what was acquired and rethrows with the old backend and every binding untouched; otherwise it commits — publishes settled ids, leaves bindings with a bind in flight holding their calls, swaps, notifies, moves the reconnect handler, and cancels the old backend's pending calls with `BackendChangedError` (stopping running Task handlers). The `unique_ptr` overload is a template on the concrete backend type and delegates to the `shared_ptr` one. | | `deregisterHandler` | `void deregisterHandler(const shared_ptr&)` | Supersedes a bind in flight (its reply releases its instance), rejects held calls with `HandlerDestroyedError`, deregisters from the active backend (if bound), resets `currentId` to 0, removes from tracking. | -| `executeVia` | `Completion executeVia(const shared_ptr&, Action, IExecutor*, function onResult = {})` | On the owner. Counts the call pending and arms the execute deadline when it is made; dispatches now when no bind is in flight, else holds it until the bind settles (a failed bind rejects it with the bind's error; a never-attached `AllowShared` binding with `"handler not bound"`). Attaches the default session. On `LocalBackend`, rejects an action whose `ActionValidator::ready` returns `false` with `morph::model::ValidationError` via `onError`, before `Model::execute` runs. Records a journal `LogEntry` for loggable actions on both success (`Outcome::Succeeded`) and a throwing `Model::execute` (`Outcome::Failed`, rethrown unchanged); a failure to serialise the result or to append the success entry happens after the mutation committed and rejects the completion with `morph::model::ActionRecordingError` instead. Dispatches through `IBackend::executeInto`, handing the backend a `detail::BridgeSink` that is simultaneously the caller's typed completion state and the backend's settle sink. The bridge-side work a result triggers (`onResult`, the subscription fan-out) runs on the owner, gated on the bridge's `CallbackToken`. A call whose handler returns a `Task` carries a `StopSource`, requested by the deadline, by a `CallbackScope` a callback was attached through, and by the backend's `cancelPending`. | +| `executeVia` | `Completion executeVia(const shared_ptr&, Action, IExecutor*, function onResult = {})` | On the owner. Counts the call pending and arms the execute deadline when it is made; dispatches now when no bind is in flight, else holds it until the bind settles (a failed bind rejects it with the bind's error; a never-attached `AllowShared` binding with `"handler not bound"`). Attaches the default session. On `LocalBackend`, rejects an action whose `ActionValidator::ready` returns `false` with `morph::model::ValidationError` via `onError`, before `Model::execute` runs. Records a journal `LogEntry` for loggable actions on both success (`Outcome::Succeeded`) and a throwing `Model::execute` (`Outcome::Failed`, rethrown unchanged); a failure to serialise the result or to append the success entry happens after the mutation committed and rejects the completion with `morph::model::ActionRecordingError` instead. Dispatches through `IBackend::executeInto`, handing the backend a `detail::BridgeSink` that is simultaneously the caller's typed completion state and the backend's settle sink. The bridge-side work a result triggers (`onResult`, the subscription fan-out) runs on the owner, gated on the bridge's `CallbackToken`. A call whose handler returns a `Task` carries a `StopSource`, requested by the deadline, by a `CallbackScope` a callback was attached through, by a `co_await` of its completion that is stopped, by the token given to `BridgeHandler::execute(action, stop)`, and by the backend's `cancelPending`. | | `setDefaultSession` | `void setDefaultSession(session::Context)` | Installs default session context; also pushes it to the active backend via `IBackend::setSession` so control envelopes (register/attach/assign/deregister) carry it too, not only `execute`. | | `defaultSession` | `session::Context defaultSession() const` | Returns snapshot of default session. | | `setExecuteDeadline` | `void setExecuteDeadline(std::chrono::milliseconds)` | Opt-in client-side execute deadline; `0` (the default) disables it. Lazily creates the backing `TimeoutScheduler`, on a private `exec::IoLoop` with one thread, on first enable. | @@ -886,6 +886,7 @@ is tolerated, as above. | ctor (custom binding) | `BridgeHandler(Bridge&, IExecutor*, shared_ptr)` | Registers the pre-built binding the same way. | | dtor | `~BridgeHandler()` | On the owner. Deregisters via `Bridge::deregisterHandler` (rejecting held calls with `HandlerDestroyedError`), but only if the bridge's `CallbackToken` is still active; a no-op if the `Bridge` was already destroyed. | | `execute` | `Completion execute(Action)` | Typed dispatch through the bridge; held until the handler's bind settles when one is in flight. For a shared handler, a payload-/result-keyed action's attach or promote step never throws synchronously — a backend refusal (e.g. `LimitPolicy::maxLiveModels`) resolves the returned `Completion` via `.onError(...)`. | +| `execute` (cancellable) | `Completion execute(Action, core::async::StopToken stop)` | As `execute(Action)`, with a caller's cancel. A stop on `stop` before the call settles rejects the `Completion` with `core::async::OperationCancelled` — posted from the thread that requested the stop — and requests stop on the call's `StopSource`, so a Task handler on a `LocalBackend` sees it; a synchronous handler runs to its end, and a remote server's handler runs on, since no cancel crosses the wire. The call stays pending until the backend settles it, as for a fired deadline. A token already stopped rejects the call without dispatching it. See the cancellation policy in [concurrency_and_lifetimes.md](../concurrency_and_lifetimes.md#cancellation-policy). | | `executeJson` | `Completion executeJson(string_view actionType, string_view bodyJson)` | Type-erased dispatch through `ActionExecuteRegistry`. | | `subscribe(cb)` | `void subscribe(function)` | Fire `cb` whenever an `R` is produced on the attached instance. | | `subscribe(scope, cb)` | `void subscribe(CallbackScope const&, function)` | As above, gated on the scope's liveness and stop state ([callback_scope.md](callback_scope.md)). Dead sinks are refused, not pruned. | diff --git a/docs/spec/core/completion.md b/docs/spec/core/completion.md index 17960b08d..f3ffe6fc1 100644 --- a/docs/spec/core/completion.md +++ b/docs/spec/core/completion.md @@ -66,7 +66,7 @@ by `std::make_shared`, which every site in the tree does | `onErrAttached` | `std::atomic` | `cbExec`, or a synchronous taker | Suppresses orphan logging; set by an `onError` attach on a state with an executor, and by `bridge::detail::takeSettled` | | `presumedOwnerThread`, `deliveredOn` | `std::atomic` | — | The debug check of an attach made outside every executor's task; see [Thread safety](#thread-safety) | | `cbExec` | `::morph::exec::IExecutor*` | set before publication | The owner; may be `nullptr` | -| `stopSource`, `stopLinks` | | `cbExec` | The call's stop source and the scope links relaying to it; released by `deliver()` | +| `stopSource`, `stopLinks` | | `cbExec` | The call's stop source, and the links relaying a stop to it — a scope's, an await's, or a caller's cancel token, which also settles the call; released by `deliver()` | **Setting a value or exception.** `setValue(T)` and `setException(exception_ptr)` are called by the producer, on any thread. The first claims `settled` with an @@ -664,16 +664,19 @@ future/promise or a monadic async type. Its scope is narrow by design: `T → U` mapping and no way to chain one asynchronous step onto another. To sequence work, the consumer must start a fresh operation from inside the handler. -- **No *work* cancellation.** There is no handle to cancel an outstanding - operation; once started, it runs to completion (or is abandoned). - `Bridge::setExecuteDeadline` (see - [Client-side execute deadline](#client-side-execute-deadline)) is not an - exception to this: it bounds how long the *caller* waits by resolving the - state early, and does nothing to the work still in flight underneath. - **Delivery**, by contrast, *can* be stopped — see - [Lifetime and stop gating](#lifetime-and-stop-gating). The distinction is - sharp and deliberate: a `CallbackScope` says "do not hand me this result", - never "stop producing it". Work-side cancellation is not offered at all. +- **No cancel API on `Completion` itself.** A `Completion` cannot cancel the + operation it reports on. The call's producer can link a cancel to it: a + `Bridge` call whose handler returns a `Task` carries a stop source, which a + stopped `CallbackScope` a callback was attached through, a stopped + `co_await`, the execute deadline and the token given to + `BridgeHandler::execute(action, stop)` all request. That token also settles + the call with `core::async::OperationCancelled`; the deadline settles it with + `ClientTimeoutError`. A synchronous handler, or a server across the wire, is + not reached by any of them: the token and the deadline settle the state + early, and the work runs on underneath (see + [Client-side execute deadline](#client-side-execute-deadline) and the + cancellation policy in + [concurrency_and_lifetimes.md](../concurrency_and_lifetimes.md#cancellation-policy)). - **Single consumer handle, but multiple handlers per outcome.** The `Completion` handle itself is move-only — only one owner at a time — but each state's `onOk`/`onErr` are vectors, so repeated `then()`/`onError()` @@ -694,9 +697,9 @@ state; the log is emitted only when the state itself is finally destroyed with a ## Out of scope -- Work cancellation — there is no mechanism to cancel an outstanding - *operation*. Stopping *delivery* of its result is - [`CallbackScope`](callback_scope.md); the two are different things. +- A cancel on the `Completion` handle — cancelling the operation belongs to + whoever produced it (see [Limitations](#limitations)). Stopping *delivery* + of its result is [`CallbackScope`](callback_scope.md). - Multiple values — `Completion` is a single-result primitive. - Synchronous blocking — there is no `wait()` or `get()`; the API is callback-only. diff --git a/include/morph/core/bridge.hpp b/include/morph/core/bridge.hpp index b803316f4..daff4afb8 100644 --- a/include/morph/core/bridge.hpp +++ b/include/morph/core/bridge.hpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -2226,6 +2227,43 @@ class BridgeHandler { } } + /// @brief Dispatches @p action as `execute(action)` does, cancellable through @p stop. + /// + /// A stop on @p stop before the call settles cancels it: the returned + /// `Completion` is rejected with `core::async::OperationCancelled`, posted + /// to the GUI executor from the thread that requested the stop, and the + /// call's stop source is asked to stop. A handler returning a `Task`, on a + /// `LocalBackend`, sees that stop at its next stop-aware `co_await`; a + /// synchronous handler runs to its end, and a remote server's handler runs + /// on, since no cancel crosses the wire. Either way the call's own outcome, + /// later, reaches nobody. A token already stopped rejects the call the same + /// way without dispatching it. + /// + /// The call stays counted as pending until the backend settles it, as a + /// call the execute deadline settled does. + /// + /// @tparam Action Concrete action type registered with `BRIDGE_REGISTER_ACTION`. + /// @param action Action to execute (moved into the dispatch). + /// @param stop The caller's cancel. Observed until the call settles; a + /// token that can never be stopped makes this `execute(action)`. + /// @return Completion that resolves on the GUI executor, or rejects with + /// `core::async::OperationCancelled` once @p stop is stopped. + template + ::morph::async::Completion::Result> execute( + Action action, ::core::async::StopToken stop) { + using R = ::morph::model::ActionTraits::Result; + if (stop.stop_requested()) { + auto [cancelled, promise] = ::morph::async::Completion::makeSettleable(_guiExec); + promise.reject(std::make_exception_ptr(::core::async::OperationCancelled{})); + return std::move(cancelled); + } + auto completion = execute(std::move(action)); + // Never empty: every path of `execute(action)` returns a completion + // built on a state, a rejected one included. + completion.state()->linkCancel(std::move(stop)); + return completion; + } + /// @brief Attaches (or re-points) this handler to the instance for @p key. /// /// Creates the instance if no live instance holds @p key, otherwise joins the diff --git a/include/morph/core/completion.hpp b/include/morph/core/completion.hpp index 18c8cffa3..7fecf43ba 100644 --- a/include/morph/core/completion.hpp +++ b/include/morph/core/completion.hpp @@ -3,6 +3,7 @@ #pragma once #include #include +#include #include #include #include @@ -145,6 +146,44 @@ struct CompletionState : std::enable_shared_from_this> { std::move(scopeStop), StopRelay{std::weak_ptr<::core::async::StopSource>{stopSource}})); } + // Relays a caller's cancel to the call: settles it with `OperationCancelled` + // first, so that outcome wins over the one the stopped work produces, then + // requests stop on its stop source. Holds both weakly. Runs on whichever + // thread requested the stop; a settle only posts its delivery. + struct CancelRelay { + std::weak_ptr state; + std::weak_ptr<::core::async::StopSource> source; + void operator()() const noexcept { + if (auto const strong = state.lock()) { + try { + strong->setException(std::make_exception_ptr(::core::async::OperationCancelled{})); + } catch (...) { // NOLINT(bugprone-empty-catch): a stop callback must not throw + // An executor that refuses the delivery leaves nothing to tell. + } + } + if (auto const strong = source.lock()) { + static_cast(strong->request_stop()); + } + } + }; + + // Links @p cancel to this call until the state is delivered: a stop on it + // settles the call with `OperationCancelled` and stops its stop source, if + // it has one. A no-op for a token that can never be stopped, a state + // already delivered, or one with no executor. A token already stopped + // cancels the call here. Runs on the owner. + void linkCancel(::core::async::StopToken cancel) { + if (cbExec == nullptr || delivered.load(std::memory_order_relaxed) || !cancel.stop_possible()) { + return; + } + checkOwner("Completion::linkCancel"); + stopLinks.push_back(std::make_shared<::core::async::StopCallback>( + std::move(cancel), CancelRelay{ + .state = this->weak_from_this(), + .source = std::weak_ptr<::core::async::StopSource>{stopSource}, + })); + } + void setValue(T val) { if (settled.exchange(true, std::memory_order_acq_rel)) { return; diff --git a/scripts/branch_partial_allowlist.json b/scripts/branch_partial_allowlist.json index 19e1fa0f1..7f801172e 100644 --- a/scripts/branch_partial_allowlist.json +++ b/scripts/branch_partial_allowlist.json @@ -109,7 +109,7 @@ }, { "file": "include/morph/core/bridge.hpp", - "line": 1772, + "line": 1773, "source": "if (_executeDeadline.count() <= 0 || !_timeoutScheduler) {", "reason": "The `!_timeoutScheduler` disjunct is unreachable by construction. `setExecuteDeadline` is the only writer of both `_executeDeadline` and `_timeoutScheduler`, and it creates `_timeoutScheduler` in the same call that makes `_executeDeadline` positive; nothing resets `_timeoutScheduler` to null afterwards (setting the deadline back to 0 only stops new calls from arming it, per that method's doc comment). So `_executeDeadline > 0 && !_timeoutScheduler` cannot hold once any positive deadline has been set." }, diff --git a/tests/test_cancellation_policy.cpp b/tests/test_cancellation_policy.cpp index 42373bb2d..47af78a47 100644 --- a/tests/test_cancellation_policy.cpp +++ b/tests/test_cancellation_policy.cpp @@ -15,7 +15,8 @@ // "Work" is the model action or server handler the verb abandons. Most cases // use a synchronous handler, which nothing can stop, so they measure what each // verb does to the callbacks and whether it waits for that work. The verbs -// whose point is to stop the work -- a stopped co_await -- are measured +// whose point is to stop the work -- a stopped co_await, a stopped execute +// token -- are measured // against a Task handler, which can observe a stop; the other verbs' Task // handler stops are measured in test_coroutine_model.cpp. @@ -458,6 +459,179 @@ TEST_CASE("Bridge::setExecuteDeadline is G0 for calls already made; a fired dead CHECK(after.ok == 0); } +TEST_CASE("execute with a stop token is G2 when stopped, and stops the Task handler running the call", + "[cancel-policy]") { + SleeperScope const sleeping; + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + Outcome outcome; + auto completion = handler.execute(CPSleep{.ms = 60'000}, stop.get_token()); + outcome.attach(completion); + REQUIRE(pumpUntil(owner, [] { return sleeper().started.load() == 1; })); + + static_cast(stop.request_stop()); + + // Not G3: the terminal callback has not run on return. + CHECK(outcome.err == 0); + // G2: OperationCancelled is posted, and runs on the next pump. + REQUIRE(pumpUntil(owner, [&] { return outcome.err == 1; })); + CHECK(holds(outcome.error)); + // The work: the Task handler is asked to stop. + REQUIRE(pumpUntil(owner, [] { return sleeper().cancelled.load() == 1; })); + CHECK(sleeper().finished.load() == 0); + // G1: the stopped handler's own outcome reaches nobody. + owner.runFor(30ms); + CHECK(outcome.ok == 0); + CHECK(outcome.err == 1); +} + +TEST_CASE("execute with a stop token stopped from another thread settles the call; a synchronous action runs on", + "[cancel-policy]") { + gate().reset(); + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + Outcome outcome; + auto completion = handler.execute(CPBlock{.tag = 1}, stop.get_token()); + outcome.attach(completion); + REQUIRE(pumpUntil(owner, [] { return gate().started.load() == 1; })); + + std::thread{[&stop] { static_cast(stop.request_stop()); }}.join(); + + CHECK(outcome.err == 0); + REQUIRE(pumpUntil(owner, [&] { return outcome.err == 1; })); + CHECK(holds(outcome.error)); + // No join, and nothing can stop a synchronous action: it is still running. + CHECK(gate().finished.load() == 0); + gate().released = true; + REQUIRE(morph::testing::waitUntil([] { return gate().finished.load() == 1; })); + owner.runFor(30ms); + CHECK(outcome.ok == 0); + CHECK(outcome.err == 1); +} + +TEST_CASE("execute with a stop token already stopped rejects the call without dispatching it", "[cancel-policy]") { + gate().reset(); + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + static_cast(stop.request_stop()); + Outcome outcome; + auto completion = handler.execute(CPBlock{.tag = 1}, stop.get_token()); + outcome.attach(completion); + + REQUIRE(pumpUntil(owner, [&] { return outcome.err == 1; })); + CHECK(holds(outcome.error)); + owner.runFor(30ms); + CHECK(gate().started.load() == 0); +} + +TEST_CASE("execute with a stop token on SimulatedRemoteBackend abandons the call; the server's handler runs on", + "[cancel-policy]") { + gate().reset(); + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + auto const server = std::make_shared(pool); + morph::bridge::Bridge bridge{std::make_unique(*server), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + REQUIRE(pumpUntil(owner, [&] { return handler.isBound(); })); + + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + Outcome outcome; + auto completion = handler.execute(CPBlock{.tag = 1}, stop.get_token()); + outcome.attach(completion); + REQUIRE(pumpUntil(owner, [] { return gate().started.load() == 1; })); + + static_cast(stop.request_stop()); + + CHECK(outcome.err == 0); + REQUIRE(pumpUntil(owner, [&] { return outcome.err == 1; })); + CHECK(holds(outcome.error)); + // Nothing crossed the wire: the server's handler finishes, and its reply + // is dropped. + CHECK(gate().finished.load() == 0); + gate().released = true; + REQUIRE(morph::testing::waitUntil([] { return gate().finished.load() == 1; })); + owner.runFor(30ms); + CHECK(outcome.ok == 0); + CHECK(outcome.err == 1); +} + +TEST_CASE("a cancel token links nothing to a call with no executor, one already delivered, or when it cannot stop", + "[cancel-policy]") { + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + + // No executor: nothing could ever deliver the cancellation. + auto const unowned = std::make_shared>(); + unowned->linkCancel(stop.get_token()); + CHECK(unowned->stopLinks.empty()); + + morph::exec::MainThreadExecutor owner; + auto [completion, promise] = morph::async::Completion::makeSettleable(&owner); + auto const state = completion.state(); + Outcome outcome; + outcome.attach(completion); + + // A token no source can stop: there is nothing to link. + state->linkCancel(core::async::StopToken{}); + CHECK(state->stopLinks.empty()); + + // Delivered: the call has nothing left to cancel, and its outcome stands. + promise.resolve(1); + REQUIRE(pumpUntil(owner, [&] { return outcome.ok == 1; })); + state->linkCancel(stop.get_token()); + CHECK(state->stopLinks.empty()); + static_cast(stop.request_stop()); + owner.runFor(20ms); + CHECK(outcome.ok == 1); + CHECK(outcome.err == 0); +} + +TEST_CASE("a cancel link that outlives its call still stops the call's stop source, and settles nothing", + "[cancel-policy]") { + morph::exec::MainThreadExecutor owner; + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource cancel; + auto const callStop = std::make_shared(); + std::shared_ptr link; + std::weak_ptr> gone; + { + auto [completion, promise] = morph::async::Completion::makeSettleable(&owner); + auto const state = completion.state(); + state->stopSource = callStop; + state->linkCancel(cancel.get_token()); + REQUIRE(state->stopLinks.size() == 1); + // Held past the state, as a stop callback running while the state + // is released would be. + link = state->stopLinks.front(); + gone = state; + } + REQUIRE(gone.expired()); + REQUIRE(link != nullptr); + + static_cast(cancel.request_stop()); + + // The relay holds the state weakly: with it gone there is nothing to + // settle, and the call's stop source is still asked to stop. + CHECK(callStop->stop_requested()); + owner.runFor(20ms); +} + namespace { /// The binding type `SubscriptionRegistry` matches on: only its instance id. From 974b62b08cf36eef9fb2f7c1805b3d6998936f4c Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Sun, 4 Oct 2026 14:53:32 +0200 Subject: [PATCH 3/3] tests: keep a late-reply test's outcomes alive across the pumps that deliver into them "a promotion reply after the handler or the bridge is gone is a no-op" declared each Outcome inside the scope it was destroying, but attached ungated then/onError handlers that capture it. A reply settling inside a later runFor() is delivered into it, so whether the test read a destroyed stack object depended on whether the pool's settle landed inside the 5 ms window. clang-asan caught it once as a stack-use-after-scope at test_async_registration.cpp:556. Both outcomes now live at test scope. Co-Authored-By: Claude Opus 5.5 --- tests/test_async_registration.cpp | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/test_async_registration.cpp b/tests/test_async_registration.cpp index ce9f0f724..f8d4fa72f 100644 --- a/tests/test_async_registration.cpp +++ b/tests/test_async_registration.cpp @@ -913,20 +913,23 @@ TEST_CASE("Bridge::assignHandlerPrimary: a promotion reply after the handler or Owner owner; morph::exec::ThreadPoolExecutor pool{2}; auto async = std::make_shared(pool); + // The outcomes outlive every pump below: their handlers are not + // scope-gated, so a reply settling inside a later runFor() is delivered + // into them after the handler or the bridge that made the call is gone. + Outcome afterHandler; + Outcome afterBridge; { morph::bridge::Bridge bridge{std::make_unique(pool), owner}; bridge.switchBackend(std::static_pointer_cast(async)); { morph::bridge::BridgeHandler handler{bridge, &owner}; - Outcome outcome; - track(outcome, handler.execute(ARCreate{.initial = 3})); + track(afterHandler, handler.execute(ARCreate{.initial = 3})); REQUIRE(pumpOwnerUntil(owner, [&] { return async->pendingCount() == 1; })); } async->completeNext(); // the handler is gone owner.runFor(std::chrono::milliseconds{5}); morph::bridge::BridgeHandler handler{bridge, &owner}; - Outcome outcome; - track(outcome, handler.execute(ARCreate{.initial = 4})); + track(afterBridge, handler.execute(ARCreate{.initial = 4})); REQUIRE(pumpOwnerUntil(owner, [&] { return async->pendingCount() == 1; })); } async->completeNext(); // the bridge is gone