Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
9e57f81
core, tests: pin the Bridge behaviour its suite ran but never checked…
Yaraslaut Oct 4, 2026
5c3b3d4
tests: pin LocalBackend and SynchronousBackendAdapter behaviour the s…
Yaraslaut Oct 4, 2026
b121cf4
tests: pin rollBackShortWrite at offset zero and across repeated roll…
Yaraslaut Oct 4, 2026
250f42b
wire: drop peekCallId's redundant digit flag; pin zero digits, the ui…
Yaraslaut Oct 4, 2026
192c484
tests: pin which dispatcher closes the registration latch, and that a…
Yaraslaut Oct 4, 2026
43af70b
async: state that a TimeoutScheduler handle is never 0; pin that canc…
Yaraslaut Oct 4, 2026
2c0c17d
tests: pin that a co_await answered by its attach is not also thrown …
Yaraslaut Oct 4, 2026
8f0d59e
tests: pin the action gate's hand-off after a held leave, the pool's …
Yaraslaut Oct 4, 2026
349ffb8
tests: pin the payload shape depth limit for every nesting form, the …
Yaraslaut Oct 4, 2026
724da78
core: drop nextOpaqueId's dead initialiser; pin OpaqueIdGenerator's b…
Yaraslaut Oct 4, 2026
583084f
tests: read RemoteServer's dispatch log lines and executeLatencyMs value
Yaraslaut Oct 4, 2026
63bd4b5
tests: pin drainedWithin(0)'s immediate answer and executeTimeout's r…
Yaraslaut Oct 4, 2026
43a598d
tests: pin RemoteServer's per-connection release and the live-model c…
Yaraslaut Oct 4, 2026
f190021
tests: cover SimulatedRemoteBackend's assignPrimary, private re-bind …
Yaraslaut Oct 4, 2026
838d979
tests: satisfy clang-tidy on the new test code
Yaraslaut Oct 4, 2026
dabfdae
tests: let a parked Task handler's callback go when it finishes; outl…
Yaraslaut Oct 4, 2026
0774ae8
ci: move one branch-allowlist line hint the new bridge tests pushed down
Yaraslaut Oct 4, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion include/morph/core/bridge.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -1344,7 +1344,10 @@ class Bridge {
if (rejectIfGone(sink, bound, settled)) {
return;
}
if (bound->primary != key || bound->currentId.load() == 0U) {
// An attach that left the binding on another key must not
// dispatch there. One that kept the key but bound no
// instance is rejected by `dispatchNow` itself.
if (bound->primary != key) {
sink->settleException(bound->bindFailure ? bound->bindFailure
: notBound());
return;
Expand Down
10 changes: 5 additions & 5 deletions include/morph/core/remote.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -1588,11 +1588,11 @@ class RemoteServer : public std::enable_shared_from_this<RemoteServer> {
/// random key; guarded defensively rather than ever handed out.
/// @return A freshly-generated, non-zero, opaque `ModelId`.
[[nodiscard]] ::morph::exec::detail::ModelId nextOpaqueId() {
uint64_t id = 0;
do {
id = _idGen.permute(++_nextId);
} while (id == 0);
return ::morph::exec::detail::ModelId{id};
for (;;) {
if (uint64_t const opaque = _idGen.permute(++_nextId); opaque != 0) {
return ::morph::exec::detail::ModelId{opaque};
}
}
}

/// Everything one dispatched execute carries from `dispatchExecute` to its
Expand Down
3 changes: 2 additions & 1 deletion include/morph/core/timeout_scheduler.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,8 @@ namespace morph::async::detail {
/// thread, so it is known before the loop has armed anything.
class TimeoutScheduler {
public:
/// @brief Opaque identifier for one scheduled callback.
/// @brief Opaque identifier for one scheduled callback. Never `0`, so a
/// holder may keep `0` for "nothing scheduled" and cancel it safely.
using Handle = std::uint64_t;

/// @brief Builds the scheduler on @p loop, whose thread runs every callback.
Expand Down
5 changes: 2 additions & 3 deletions include/morph/core/wire.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -571,8 +571,8 @@ inline constexpr std::array<std::string_view, 12> kOmittableEnvelopeKeys{
while (idx < window.size() && (window[idx] == ' ' || window[idx] == '\t')) {
++idx;
}
// No digits leaves `value` at 0, which is already the "no id" answer.
std::uint64_t value = 0;
bool sawDigit = false;
for (; idx < window.size(); ++idx) {
// idx is bounded by the loop condition.
// NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-avoid-unchecked-container-access)
Expand All @@ -585,9 +585,8 @@ inline constexpr std::array<std::string_view, 12> kOmittableEnvelopeKeys{
return 0; // out of range β€” treat as unrecoverable rather than wrap
}
value = (value * 10U) + digit;
sawDigit = true;
}
return sawDigit ? value : 0;
return value;
}

} // namespace detail
Expand Down
2 changes: 1 addition & 1 deletion scripts/branch_partial_allowlist.json
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@
},
{
"file": "include/morph/core/bridge.hpp",
"line": 1773,
"line": 1776,
"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."
},
Expand Down
50 changes: 50 additions & 0 deletions tests/test_async_delay.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,11 @@
#include <core/async/Cancellation.hpp>
#include <core/async/StopToken.hpp>
#include <core/async/Task.hpp>
#include <cstddef>
#include <memory>
#include <morph/core/coroutine.hpp>
#include <morph/core/executor.hpp>
#include <morph/core/io_loop.hpp>
#include <morph/core/timeout_scheduler.hpp>
#include <thread>

Expand Down Expand Up @@ -148,3 +150,51 @@ TEST_CASE("a delay whose coroutine is destroyed while it waits withdraws its tim
REQUIRE(morph::testing::waitUntil([&] { return past.load(); }));
REQUIRE_FALSE(seen->finished.load());
}

namespace {

// Timers armed on @p loop's event loop, read on the loop's own thread -- which
// also means every schedule or cancel posted to it before has been applied.
std::size_t armedTimers(morph::exec::IoLoop& loop) {
std::size_t count = 0;
loop.runAndWait([&loop, &count] { count = loop.loop().pendingTimerCount(); });
return count;
}

} // namespace

TEST_CASE("a stop that cancels a delay also retires its timer", "[coroutine][delay]") {
morph::exec::IoLoop loop;
morph::async::detail::TimeoutScheduler scheduler{loop};
morph::exec::MainThreadExecutor exec;
auto seen = std::make_shared<DelayObserved>();
auto const before = armedTimers(loop);

auto task = waitFor(&scheduler, 1h, seen);
// NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const in the standard.
core::async::StopSource stop;
task.handle().promise().setStopToken(stop.get_token());
morph::async::spawn(exec, std::move(task));
REQUIRE(pumpUntil(exec, [&] { return seen->reachedAwait.load(); }));
REQUIRE(armedTimers(loop) == before + 1);

stop.request_stop();
REQUIRE(pumpUntil(exec, [&] { return seen->finished.load(); }));
REQUIRE(seen->cancelled);
CHECK(armedTimers(loop) == before);
}

TEST_CASE("a delay whose coroutine is destroyed while it waits retires its timer", "[coroutine][delay]") {
morph::exec::IoLoop loop;
morph::async::detail::TimeoutScheduler scheduler{loop};
auto seen = std::make_shared<DelayObserved>();
auto const before = armedTimers(loop);
{
auto task = waitFor(&scheduler, 1h, seen);
task.handle().resume();
REQUIRE(seen->reachedAwait.load());
REQUIRE(armedTimers(loop) == before + 1);
}
CHECK(armedTimers(loop) == before);
CHECK_FALSE(seen->finished.load());
}
146 changes: 146 additions & 0 deletions tests/test_backend_extra.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
#include <mutex>
#include <stdexcept>
#include <thread>
#include <utility>
#include <vector>

#include "test_support.hpp"
Expand Down Expand Up @@ -432,3 +433,148 @@ TEST_CASE("morph::backend::LocalBackend: amortised pending compaction bounds the

CHECK(cancelledCount == kRounds);
}

TEST_CASE("morph::backend::LocalBackend: trackedPendingCount counts every call still in flight",
"[backend][local][pending]") {
// Declared before the pool and the backend: the parked ops read it until
// the backend's teardown has drained them.
std::atomic<bool> release{false};
std::atomic<int> ran{0};
morph::exec::ThreadPoolExecutor pool{2};
SyncExecutor cbExec;
morph::backend::LocalBackend backend{pool};
auto mid = backend.registerModel("BE_CounterModel", morph::model::detail::ModelFactory::create<CounterModel>);
CHECK(backend.trackedPendingCount() == 0);

std::vector<morph::async::Completion<std::shared_ptr<void>>> live;
live.reserve(3);
for (int i = 0; i < 3; ++i) {
live.push_back(backend.execute(mid, pendingCall([&release, &ran] {
while (!release.load(std::memory_order_acquire)) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
ran.fetch_add(1);
}),
&cbExec));
}
CHECK(backend.trackedPendingCount() == 3);

release.store(true, std::memory_order_release);
REQUIRE(morph::testing::waitUntil([&] { return ran.load() == 3; }));
}

TEST_CASE("morph::backend::LocalBackend: executeLatencyMs covers the time the handler ran",
"[backend][local][observability]") {
morph::observe::ScopedObserveOverride const guard;
morph::exec::ThreadPoolExecutor pool{2};
morph::exec::MainThreadExecutor cbExec;
morph::backend::LocalBackend backend{pool};
auto mid = backend.registerModel("BE_CounterModel", morph::model::detail::ModelFactory::create<CounterModel>);

std::mutex sampleMtx;
std::vector<double> latencies;
morph::observe::setMetricSink([&](const morph::observe::MetricEvent& evt) {
if (evt.metric == morph::observe::Metric::executeLatencyMs) {
std::scoped_lock const lock{sampleMtx};
latencies.push_back(evt.value);
}
});

std::atomic<bool> done{false};
backend.execute(mid, pendingCall([] { std::this_thread::sleep_for(std::chrono::milliseconds(20)); }), &cbExec)
.then([&](const std::shared_ptr<void>&) { done = true; });
REQUIRE(morph::testing::pumpOwnerUntil(cbExec, [&] { return done.load(); }));

std::scoped_lock const lock{sampleMtx};
REQUIRE(latencies.size() == 1);
// A lower bound only: the handler slept 20 ms inside the measured span,
// and how much longer the span took depends on the machine.
CHECK(latencies.front() >= 20.0);
}

namespace {

/// What a hand-built Task handler kept when it started: the token it was given
/// and the callback that ends it.
struct ParkedTaskHandler {
std::atomic<bool> started{false};
core::async::StopToken token;
morph::backend::detail::ActionCall::LocalDone done;

/// Ends the handler. The callback is taken out before it runs: it holds the
/// call, the call holds this handler, and leaving it stored keeps both alive.
void finish(std::exception_ptr error) { std::exchange(done, nullptr)(nullptr, std::move(error)); }
};

/// A Task-handler call that starts and then stays suspended: it records its
/// token and its completion callback in the `ParkedTaskHandler` the call
/// carries, and returns without finishing.
morph::backend::detail::ActionCall parkedTaskCall(const std::shared_ptr<ParkedTaskHandler>& slot) {
morph::backend::detail::ActionCall call;
call.modelTypeId = "BE_CounterModel";
call.actionTypeId = "BE_CounterAction";
call.action = slot;
// NOLINTNEXTLINE(performance-unnecessary-value-param): ActionCall::localOpAsync fixes the by-value signature
call.localOpAsync = [](morph::model::detail::IModelHolder& /*holder*/, std::shared_ptr<void> action,
const std::shared_ptr<morph::exec::detail::TaskResumer>& /*executor*/,
core::async::StopToken token, morph::backend::detail::ActionCall::LocalDone done) {
auto* parked = static_cast<ParkedTaskHandler*>(action.get());
parked->token = std::move(token);
parked->done = std::move(done);
parked->started.store(true);
};
return call;
}

} // namespace

TEST_CASE("morph::backend::LocalBackend: destruction requests stop on a Task handler still running",
"[backend][local][lifetime]") {
auto slot = std::make_shared<ParkedTaskHandler>();
morph::exec::ThreadPoolExecutor pool{2};
SyncExecutor cbExec;
{
morph::backend::LocalBackend backend{pool};
auto mid = backend.registerModel("BE_CounterModel", morph::model::detail::ModelFactory::create<CounterModel>);
auto pending = backend.execute(mid, parkedTaskCall(slot), &cbExec);
REQUIRE(morph::testing::waitUntil([&] { return slot->started.load(); }));
CHECK_FALSE(slot->token.stop_requested());
}
// Nothing cancelled the call before the backend went: only its destructor
// could have asked the handler to stop.
CHECK(slot->token.stop_requested());

// The handler unwinds after its backend is gone, as a stopped one would.
slot->finish(std::make_exception_ptr(std::runtime_error{"stopped"}));
}

TEST_CASE("morph::backend::LocalBackend: a call cancelled while it waited behind a Task handler finishes as a failure",
"[backend][local][observability]") {
morph::observe::ScopedObserveOverride const guard;
std::atomic<int> errors{0};
morph::observe::setMetricSink([&errors](const morph::observe::MetricEvent& evt) {
if (evt.metric == morph::observe::Metric::executeErrors) {
errors.fetch_add(1);
}
});
auto slot = std::make_shared<ParkedTaskHandler>();
std::atomic<bool> queuedRan{false};
morph::exec::ThreadPoolExecutor pool{2};
SyncExecutor cbExec;
morph::backend::LocalBackend backend{pool};
auto mid = backend.registerModel("BE_CounterModel", morph::model::detail::ModelFactory::create<CounterModel>);

// The Task handler holds the instance's action gate; the ordinary call
// waits behind it, and is failed there by the cancellation.
auto task = backend.execute(mid, parkedTaskCall(slot), &cbExec);
REQUIRE(morph::testing::waitUntil([&] { return slot->started.load(); }));
auto queued = backend.execute(mid, pendingCall([&queuedRan] { queuedRan.store(true); }), &cbExec);
backend.cancelPending(std::make_exception_ptr(morph::backend::BackendChangedError{}));
REQUIRE(errors.load() == 0);

// The Task handler finishes successfully, handing the gate to the waiting
// call: it is recorded as the failure `cancelPending` made it, not run.
slot->finish(nullptr);
REQUIRE(morph::testing::waitUntil([&] { return errors.load() == 1; }));
CHECK_FALSE(queuedRan.load());
}
46 changes: 46 additions & 0 deletions tests/test_backend_registration_surface.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
#include <condition_variable>
#include <exception>
#include <functional>
#include <future>
#include <memory>
#include <morph/core/backend.hpp>
#include <morph/core/bridge.hpp>
Expand Down Expand Up @@ -784,3 +785,48 @@ TEST_CASE(
CHECK(errRan.load() == 2);
CHECK(inner->cancels.load() == 1);
}

// ── A synchronous verb re-entered from the control strand ───────────────────

namespace {

/// @brief A wrapped backend whose `listInstances("outer")` calls back into the
/// adapter's own synchronous `listInstances` β€” from inside the strand
/// task the adapter is running it in.
struct ReentrantBackend : RecordingBackend {
SynchronousBackendAdapter* adapter = nullptr;
std::vector<std::string> nested;

std::vector<std::string> listInstances(const std::string& typeId) override {
if (typeId == "outer" && adapter != nullptr) {
nested = adapter->listInstances("inner");
}
return {typeId};
}
};

} // namespace

TEST_CASE("morph::backend::SynchronousBackendAdapter: a synchronous verb called on its own strand runs inline",
"[backend][registration-surface][threading]") {
// Waiting for the strand from inside the strand would never return, so the
// outer call is made from a detached thread and waited for with a budget:
// a deadlock fails the case instead of hanging the suite. The adapter and
// its executor are leaked on that path, since both have a thread stuck in
// them that teardown would wait for.
auto pool = std::make_unique<morph::exec::ThreadPoolExecutor>(1);
auto inner = std::make_shared<ReentrantBackend>();
auto adapter = std::make_unique<SynchronousBackendAdapter>(inner, *pool);
inner->adapter = adapter.get();

auto outcome = std::make_shared<std::promise<std::vector<std::string>>>();
auto answer = outcome->get_future();
std::thread{[outer = adapter.get(), outcome] { outcome->set_value(outer->listInstances("outer")); }}.detach();
if (answer.wait_for(morph::testing::kDefaultWaitBudget) != std::future_status::ready) {
[[maybe_unused]] auto const* const leakedAdapter = adapter.release();
[[maybe_unused]] auto const* const leakedPool = pool.release();
FAIL("listInstances re-entered from the control strand never returned");
}
REQUIRE(answer.get() == std::vector<std::string>{"outer"});
REQUIRE(inner->nested == std::vector<std::string>{"inner"});
}
Loading
Loading