diff --git a/include/morph/core/bridge.hpp b/include/morph/core/bridge.hpp index daff4afb8..460f76cde 100644 --- a/include/morph/core/bridge.hpp +++ b/include/morph/core/bridge.hpp @@ -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; diff --git a/include/morph/core/remote.hpp b/include/morph/core/remote.hpp index 85082ebc2..aa730fd85 100644 --- a/include/morph/core/remote.hpp +++ b/include/morph/core/remote.hpp @@ -1588,11 +1588,11 @@ class RemoteServer : public std::enable_shared_from_this { /// 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 diff --git a/include/morph/core/timeout_scheduler.hpp b/include/morph/core/timeout_scheduler.hpp index a6507e7b8..39e3296cf 100644 --- a/include/morph/core/timeout_scheduler.hpp +++ b/include/morph/core/timeout_scheduler.hpp @@ -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. diff --git a/include/morph/core/wire.hpp b/include/morph/core/wire.hpp index 71e14a530..977281012 100644 --- a/include/morph/core/wire.hpp +++ b/include/morph/core/wire.hpp @@ -571,8 +571,8 @@ inline constexpr std::array 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) @@ -585,9 +585,8 @@ inline constexpr std::array 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 diff --git a/scripts/branch_partial_allowlist.json b/scripts/branch_partial_allowlist.json index 7f801172e..c255f19a3 100644 --- a/scripts/branch_partial_allowlist.json +++ b/scripts/branch_partial_allowlist.json @@ -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." }, diff --git a/tests/test_async_delay.cpp b/tests/test_async_delay.cpp index 238893f63..1437165e7 100644 --- a/tests/test_async_delay.cpp +++ b/tests/test_async_delay.cpp @@ -9,9 +9,11 @@ #include #include #include +#include #include #include #include +#include #include #include @@ -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(); + 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(); + 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()); +} diff --git a/tests/test_backend_extra.cpp b/tests/test_backend_extra.cpp index 01c4f15f8..04144d1bb 100644 --- a/tests/test_backend_extra.cpp +++ b/tests/test_backend_extra.cpp @@ -14,6 +14,7 @@ #include #include #include +#include #include #include "test_support.hpp" @@ -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 release{false}; + std::atomic 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); + CHECK(backend.trackedPendingCount() == 0); + + std::vector>> 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); + + std::mutex sampleMtx; + std::vector 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 done{false}; + backend.execute(mid, pendingCall([] { std::this_thread::sleep_for(std::chrono::milliseconds(20)); }), &cbExec) + .then([&](const std::shared_ptr&) { 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 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& 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 action, + const std::shared_ptr& /*executor*/, + core::async::StopToken token, morph::backend::detail::ActionCall::LocalDone done) { + auto* parked = static_cast(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(); + morph::exec::ThreadPoolExecutor pool{2}; + SyncExecutor cbExec; + { + morph::backend::LocalBackend backend{pool}; + auto mid = backend.registerModel("BE_CounterModel", morph::model::detail::ModelFactory::create); + 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 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(); + std::atomic 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); + + // 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()); +} diff --git a/tests/test_backend_registration_surface.cpp b/tests/test_backend_registration_surface.cpp index 50754c374..9ea2b4583 100644 --- a/tests/test_backend_registration_surface.cpp +++ b/tests/test_backend_registration_surface.cpp @@ -25,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -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 nested; + + std::vector 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(1); + auto inner = std::make_shared(); + auto adapter = std::make_unique(inner, *pool); + inner->adapter = adapter.get(); + + auto outcome = std::make_shared>>(); + 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{"outer"}); + REQUIRE(inner->nested == std::vector{"inner"}); +} diff --git a/tests/test_bridge_bind_paths.cpp b/tests/test_bridge_bind_paths.cpp index 3453bb787..39bb0d155 100644 --- a/tests/test_bridge_bind_paths.cpp +++ b/tests/test_bridge_bind_paths.cpp @@ -6,6 +6,7 @@ // follows. Every test drives a backend whose bind and promote replies the test // settles by hand, so each ordering is exact rather than a race. +#include #include #include #include @@ -16,11 +17,15 @@ #include #include #include +#include #include #include +#include #include #include #include +#include +#include #include #include @@ -328,6 +333,9 @@ TEST_CASE("Bridge destruction rejects a call still waiting for its bind", "[brid recordFailure(bridge->executeVia(binding, BbpPlain{.value = 1}, &owner), failure); bridge.reset(); REQUIRE(pumpOwnerUntil(owner, [&] { return failure.settled; })); + // The bridge's own error, not the one a destroyed handler gives: the + // binding is still alive, so the reason is that the bridge went away. + REQUIRE(failure.what == morph::backend::BridgeDestroyedError{}.what()); } TEST_CASE("A bind reply landing after the bridge is gone releases its instance on a backend the caller kept", @@ -529,7 +537,16 @@ TEST_CASE("A promote that lands after an attach already keyed the binding keeps REQUIRE(rig.bridge->bindingPrimary(handler.binding()) == "9"); } -TEST_CASE("A failed promote leaves the binding unkeyed", "[bridge][promote]") { +TEST_CASE("A failed promote leaves the binding unkeyed and logs the backend's reason", "[bridge][promote]") { + std::mutex logMtx; + std::vector logged; + morph::log::ScopedLoggerOverride const guard{ + [&](morph::log::LogLevel, std::string_view msg) { + std::scoped_lock const lock{logMtx}; + logged.emplace_back(msg); + }, + morph::log::LogLevel::error, + }; Rig rig; rig.gate->bindMode = Reply::Inline; BridgeHandler handler{*rig.bridge, &rig.owner}; @@ -537,7 +554,12 @@ TEST_CASE("A failed promote leaves the binding unkeyed", "[bridge][promote]") { handler.execute(BbpMake{}).then([&](const BbpMade&) { done.store(true); }); REQUIRE(pumpOwnerUntil(rig.owner, [&] { return done.load(); })); rig.gate->rejectPromote(); - rig.owner.runFor(std::chrono::milliseconds{5}); + REQUIRE(pumpOwnerUntil(rig.owner, [&] { + std::scoped_lock const lock{logMtx}; + return std::ranges::any_of(logged, [](const std::string& line) { + return line.contains("promotion of 'BBP_Model' failed: promote refused"); + }); + })); REQUIRE(rig.bridge->bindingPrimary(handler.binding()).empty()); } @@ -562,6 +584,19 @@ TEST_CASE("A keyed call whose bind is refused is rejected with the bind's error" REQUIRE(failure.what == "attach refused"); } +TEST_CASE("A keyed call whose attach settles with no instance is rejected as not bound", "[bridge][bind]") { + Rig rig; + BridgeHandler handler{*rig.bridge, &rig.owner}; + Failure failure; + recordFailure(handler.execute(BbpLoad{.id = 3}), failure); + REQUIRE(rig.gate->heldBinds() == 1); + + GateBackend::resolveFirst(*rig.gate->bindStash(), ModelId{}); + REQUIRE(pumpOwnerUntil(rig.owner, [&] { return failure.settled; })); + REQUIRE(failure.what == "handler not bound"); + REQUIRE(rig.bridge->pendingCalls() == 0); +} + // ── Binds that never produce an outcome, and the calls queued behind them ──── TEST_CASE("A backend that returns an empty bind completion fails the bind", "[bridge][bind]") { @@ -735,6 +770,58 @@ TEST_CASE("A handler registration posted to the owner is dropped when the bridge SUCCEED(); } +// A superseded reply must stay superseded however many binds follow it: the +// count that marks it stale must never come back round to the value it carries. + +TEST_CASE("An attach superseded by a switch stays superseded once a newer attach is issued", + "[bridge][bind][switch]") { + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + auto old = std::make_shared(pool); + Bridge bridge{std::make_unique(pool), owner}; + bridge.switchBackend(std::static_pointer_cast(old)); + BridgeHandler handler{bridge, &owner}; + handler.attach(1); + REQUIRE(old->heldBinds() == 1); + + auto next = std::make_unique(pool); + auto* nextGate = next.get(); + bridge.switchBackend(std::move(next)); + handler.attach(2); + REQUIRE(nextGate->heldBinds() == 1); + + // The first attach's reply lands last: it is released, and binds nothing. + old->resolveBind(); + owner.runFor(std::chrono::milliseconds{5}); + REQUIRE(*old->releasedCounter() == 1); + REQUIRE_FALSE(handler.isBound()); + REQUIRE(bridge.bindingPrimary(handler.binding()).empty()); +} + +TEST_CASE("A bind superseded by a switch stays superseded once a reconnect issues another", + "[bridge][bind][switch][reconnect]") { + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + auto old = std::make_shared(pool); + Bridge bridge{std::make_unique(pool), owner}; + bridge.switchBackend(std::static_pointer_cast(old)); + BridgeHandler const handler{bridge, &owner}; + REQUIRE(old->heldBinds() == 1); + + auto next = std::make_unique(pool); + auto* nextGate = next.get(); + bridge.switchBackend(std::move(next)); + REQUIRE(nextGate->heldBinds() == 1); + nextGate->fireReconnect(); + REQUIRE(pumpOwnerUntil(owner, [&] { return nextGate->heldBinds() == 2; })); + + // The first bind's reply lands last: it is released, and binds nothing. + old->resolveBind(); + owner.runFor(std::chrono::milliseconds{5}); + REQUIRE(*old->releasedCounter() == 1); + REQUIRE_FALSE(handler.isBound()); +} + TEST_CASE("A superseded bind that replies with no instance releases nothing", "[bridge][bind][switch]") { morph::exec::ThreadPoolExecutor pool{2}; morph::exec::MainThreadExecutor owner; @@ -752,3 +839,78 @@ TEST_CASE("A superseded bind that replies with no instance releases nothing", "[ owner.runFor(std::chrono::milliseconds{5}); REQUIRE(*old->releasedCounter() == 0); } + +// ── What a rejected or released operation leaves behind ────────────────────── + +TEST_CASE("An attach held behind a bind is dropped, not issued, when its handler is destroyed", + "[bridge][bind][teardown]") { + Rig rig; + // Outlives the handler: the keyed call is rejected with its destruction, + // and that failure is delivered on the owner afterwards. + Failure keyed; + { + BridgeHandler handler{*rig.bridge, &rig.owner}; + recordFailure(handler.execute(BbpLoad{.id = 1}), keyed); + REQUIRE(rig.gate->heldBinds() == 1); + handler.attach(2); // waits behind the keyed call's bind + } + rig.owner.runFor(std::chrono::milliseconds{5}); + // Only the keyed call's bind was ever issued: the attach was rejected with + // the handler's destruction rather than run against a deregistered binding. + REQUIRE(rig.gate->heldBinds() == 1); +} + +TEST_CASE("Bridge::assignHandlerPrimary with an empty key promotes nothing", "[bridge][promote]") { + Rig rig; + rig.gate->bindMode = Reply::Inline; + auto binding = rig.bridge->registerHandler(); + REQUIRE(Bridge::isBound(binding)); + + rig.bridge->assignHandlerPrimary(binding, ""); + REQUIRE(rig.gate->heldPromotes() == 0); + REQUIRE(rig.bridge->bindingPrimary(binding).empty()); +} + +TEST_CASE("Destroying one handler leaves the bridge tracking the others", "[bridge][bind][switch]") { + Rig rig; + rig.gate->bindMode = Reply::Inline; + BridgeHandler const survivor{*rig.bridge, &rig.owner}; + { + BridgeHandler const dropped{*rig.bridge, &rig.owner}; + } + + // A switch re-binds every handler the bridge still tracks: the survivor, + // and only the survivor. + auto next = std::make_unique(rig.pool); + auto* nextGate = next.get(); + rig.bridge->switchBackend(std::move(next)); + REQUIRE(nextGate->heldBinds() == 1); +} + +TEST_CASE("A call made on the owner before a posted handler registration runs is held, not refused", + "[bridge][bind]") { + Rig rig; + rig.gate->bindMode = Reply::Inline; + + // Constructed off the owner, so its registration is posted there. + std::unique_ptr> handler; + std::thread constructor{[&] { handler = std::make_unique>(*rig.bridge, &rig.owner); }}; + constructor.join(); + REQUIRE_FALSE(handler->isBound()); + + std::atomic result{-1}; + Failure failure; + handler->execute(BbpPlain{.value = 3}) + .then([&](const BbpValue& val) { result.store(val.value); }) + .onError([&](const std::exception_ptr& err) { + try { + std::rethrow_exception(err); + } catch (const std::exception& exc) { + failure.what = exc.what(); + } + failure.settled = true; + }); + REQUIRE(pumpOwnerUntil(rig.owner, [&] { return result.load() != -1 || failure.settled; })); + REQUIRE(failure.what.empty()); + REQUIRE(result.load() == 3); +} diff --git a/tests/test_bridge_owner.cpp b/tests/test_bridge_owner.cpp index 5052fc93d..bf9971c0e 100644 --- a/tests/test_bridge_owner.cpp +++ b/tests/test_bridge_owner.cpp @@ -302,6 +302,23 @@ TEST_CASE("Bridge: the subscription fan-out of a result settled on a pool thread REQUIRE(recorder.allPosted("Bridge::publishResult")); } +TEST_CASE("Bridge: a result the backend settles on the owner is published before execute returns", + "[bridge][owner][subscription]") { + // A backend whose strands run inline settles on the calling thread, which + // is the owner here: the bridge-side work runs there at once rather than + // as a later owner task. The owner is never pumped. + morph::testing::InlineExecutor inlinePool; + morph::testing::InlineExecutor gui; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(inlinePool), owner}; + morph::bridge::BridgeHandler handler{bridge, &gui}; + + int seen = 0; + handler.subscribe([&](Pong pong) { seen = pong.value; }); + handler.execute(Ping{.value = 3}); + REQUIRE(seen == 6); +} + // ── Owner-only verbs ───────────────────────────────────────────────────────── TEST_CASE("Bridge: owner-only verbs run inside the owner's task, and are caught off it", "[bridge][owner]") { @@ -393,6 +410,29 @@ TEST_CASE("BridgeHandler: teardown runs on the owner, and a destruction off it i } } +#ifndef NDEBUG +TEST_CASE("BridgeHandler: a guiExec other than the owner is checked once, where it runs its tasks", + "[bridge][owner]") { + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + OwnerProbeRecorder const recorder{owner.coreExecutor()}; + + SECTION("the owner itself needs no check") { + morph::bridge::BridgeHandler const handler{bridge, &owner}; + owner.runFor(std::chrono::milliseconds{5}); + REQUIRE(recorder.count("BridgeHandler::guiExec") == 0); + } + + SECTION("another executor is checked inside one of its own tasks") { + morph::exec::MainThreadExecutor gui; + morph::bridge::BridgeHandler const handler{bridge, &gui}; + REQUIRE(pumpOwnerUntil(gui, [&] { return recorder.count("BridgeHandler::guiExec") == 1; })); + REQUIRE_FALSE(recorder.at("BridgeHandler::guiExec").front().onOwner); + } +} +#endif + // ── Backends below the bridge ──────────────────────────────────────────────── TEST_CASE("SynchronousBackendAdapter: a bind registered from inside an action is issued on the owner", diff --git a/tests/test_bridge_remote.cpp b/tests/test_bridge_remote.cpp index c9bdaa6a0..8a880da00 100644 --- a/tests/test_bridge_remote.cpp +++ b/tests/test_bridge_remote.cpp @@ -8,7 +8,9 @@ #include #include #include +#include #include +#include #include "test_support.hpp" @@ -96,3 +98,99 @@ TEST_CASE("morph::backend::SimulatedRemoteBackend: multiple actions on same hand REQUIRE(count.load() == numActions); REQUIRE(sum.load() == 15); // 1+2+3+4+5 } + +TEST_CASE("morph::backend::SimulatedRemoteBackend: assignPrimary files a live instance under its key", + "[bridge][remote]") { + morph::exec::ThreadPoolExecutor serverPool{2}; + auto const server = std::make_shared(serverPool); + morph::backend::SimulatedRemoteBackend backend{*server}; + + auto const mid = backend.registerModel("TestR_EchoModel", {}); + REQUIRE(mid.v != 0U); + REQUIRE(backend.listInstances("TestR_EchoModel").empty()); + + backend.assignPrimary(mid, "TestR_EchoModel", "filed-key"); + CHECK(backend.listInstances("TestR_EchoModel") == std::vector{"filed-key"}); +} + +TEST_CASE("morph::backend::SimulatedRemoteBackend: binding a private instance releases the one it replaces", + "[bridge][remote]") { + morph::exec::ThreadPoolExecutor serverPool{2}; + auto const server = std::make_shared(serverPool); + morph::backend::SimulatedRemoteBackend backend{*server}; + auto const liveModels = [&server] { + return morph::testing::awaitAnswer([&server](auto& owner) { return server->health(owner); }).liveModels; + }; + + auto const old = backend.registerModel("TestR_EchoModel", {}); + REQUIRE(liveModels() == 1U); + + morph::backend::detail::BindRequest request; + request.typeId = "TestR_EchoModel"; + request.current = old; + morph::exec::MainThreadExecutor cbExec; + auto const bound = morph::testing::awaitValueOn(cbExec, backend.bindModel(std::move(request), cbExec)); + CHECK(bound.v != old.v); + CHECK(liveModels() == 1U); +} + +namespace { +std::atomic& remoteBlockRelease() { + static std::atomic value{false}; + return value; +} +std::atomic& remoteBlockStarted() { + static std::atomic value{0}; + return value; +} +} // namespace + +struct BlockAction {}; +struct BlockModel { + int execute(BlockAction /*action*/) { + remoteBlockStarted().fetch_add(1); + while (!remoteBlockRelease().load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + return 0; + } +}; + +BRIDGE_REGISTER_MODEL(BlockModel, "TestR_BlockModel") +// The generated fromJson body declares a non-const local inside the macro. +// NOLINTNEXTLINE(misc-const-correctness) +BRIDGE_REGISTER_ACTION(BlockModel, BlockAction, "TestR_BlockAction") + +TEST_CASE("morph::backend::SimulatedRemoteBackend: cancelPending fails every call still in flight", + "[bridge][remote]") { + remoteBlockRelease().store(false); + remoteBlockStarted().store(0); + morph::exec::ThreadPoolExecutor serverPool{2}; + auto const server = std::make_shared(serverPool); + morph::exec::MainThreadExecutor cbExec; + auto owned = std::make_unique(*server); + auto* const backend = owned.get(); + morph::bridge::Bridge bridge{std::move(owned), cbExec}; + morph::bridge::BridgeHandler handler{bridge, &cbExec}; + + // Three calls on one instance: the first runs and blocks, the other two + // queue behind it, so all three are in flight together. + constexpr int kCalls = 3; + std::atomic errors{0}; + for (int idx = 0; idx < kCalls; ++idx) { + handler.execute(BlockAction{}).then([](int) {}).onError([&errors](const std::exception_ptr&) { + errors.fetch_add(1); + }); + } + REQUIRE(morph::testing::waitUntil([] { return remoteBlockStarted().load() == 1; })); + + backend->cancelPending(std::make_exception_ptr(std::runtime_error("cancelled"))); + for (int idx = 0; idx < 100 && errors.load() < kCalls; ++idx) { + cbExec.runFor(std::chrono::milliseconds(10)); + } + CHECK(errors.load() == kCalls); + + remoteBlockRelease().store(true); + REQUIRE(morph::testing::awaitAnswer( + [&server](auto& owner) { return server->drainedWithin(std::chrono::milliseconds{5000}, owner); })); +} diff --git a/tests/test_client_execute_deadline.cpp b/tests/test_client_execute_deadline.cpp index 0a4ee0b15..810833681 100644 --- a/tests/test_client_execute_deadline.cpp +++ b/tests/test_client_execute_deadline.cpp @@ -13,6 +13,9 @@ #include #include +#include +#include +#include #include #include #include @@ -399,3 +402,52 @@ TEST_CASE( // deadline. CHECK(stateWatch.expired()); } + +TEST_CASE("Bridge::setExecuteDeadline(0) after a positive deadline disarms it for calls made afterwards", + "[core][bridge][client-deadline]") { + // The scheduler a positive deadline created stays for the bridge's + // lifetime; disabling the deadline must still leave new calls unarmed + // rather than arming them with a zero-length one. + morph::exec::MainThreadExecutor exec; + morph::bridge::Bridge bridge{std::make_unique(), exec}; + bridge.setExecuteDeadline(std::chrono::milliseconds{50}); + bridge.setExecuteDeadline(std::chrono::milliseconds{0}); + morph::bridge::BridgeHandler handler{bridge, &exec}; + + bool resolved = false; + handler.execute(DeadlineCount{.x = 1}) + .then([&resolved](int) { resolved = true; }) + .onError([&resolved](const std::exception_ptr&) { resolved = true; }); + exec.runFor(std::chrono::milliseconds{200}); + CHECK_FALSE(resolved); +} + +#ifdef __linux__ +namespace { + +/// Threads in this process, one directory entry each under /proc/self/task. +std::size_t liveThreadCount() { + std::size_t count = 0; + for ([[maybe_unused]] auto const& entry : std::filesystem::directory_iterator{"/proc/self/task"}) { + ++count; + } + return count; +} + +} // namespace + +TEST_CASE("Bridge::setExecuteDeadline(0) starts no timer thread", "[core][bridge][client-deadline]") { + // The scheduler, and the one thread its loop owns, are created by the + // first call that enables a deadline -- never by one that leaves it off. + // `<=` rather than `==`: a thread from an earlier case still exiting can + // only lower the second count, never raise it. + morph::exec::MainThreadExecutor exec; + morph::bridge::Bridge bridge{std::make_unique(), exec}; + auto const before = liveThreadCount(); + bridge.setExecuteDeadline(std::chrono::milliseconds{0}); + CHECK(liveThreadCount() <= before); + + bridge.setExecuteDeadline(std::chrono::milliseconds{50}); + CHECK(liveThreadCount() > before); +} +#endif // __linux__ diff --git a/tests/test_completion_owner.cpp b/tests/test_completion_owner.cpp index 27ba932f0..1bcc35cda 100644 --- a/tests/test_completion_owner.cpp +++ b/tests/test_completion_owner.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include #include @@ -286,3 +287,21 @@ TEST_CASE("Completion: an owner that does not run one task at a time is refused CHECK(recorder.count("Completion: an owner that is not serial") == 0U); } } + +TEST_CASE("OwnedState::ask from off the owner posts its body to the owner", "[completion][owner]") { + morph::exec::MainThreadExecutor owner; + morph::exec::ThreadPoolExecutor pool{1}; + morph::exec::detail::OwnedState owned{owner, std::make_shared(3)}; + + Completion answer; + onPool(pool, [&] { + answer = owned.ask("OwnedStateTest::ask", owner, + [&owner](int& state) { return state == 3 && morph::exec::runningOn(owner); }); + }); + + std::optional ranOnOwner; + owner.post([&] { answer.then([&](bool value) { ranOnOwner = value; }); }); + drainCounting(owner); + REQUIRE(ranOnOwner.has_value()); + CHECK(*ranOnOwner); +} diff --git a/tests/test_coroutine_client.cpp b/tests/test_coroutine_client.cpp index d31695242..f1967114f 100644 --- a/tests/test_coroutine_client.cpp +++ b/tests/test_coroutine_client.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -389,3 +390,54 @@ TEST_CASE("co_await whose handler cannot be attached rethrows at the co_await an REQUIRE_FALSE(seen->value.has_value()); REQUIRE_FALSE(seen->cancelled); } + +namespace { + +/// Runs every task inline, then -- once armed -- reports the outermost post as +/// failed anyway, after the task has run: a queue that accepted the work and +/// then threw on its way out. +class RunsThenThrowsExecutor : public morph::exec::IExecutor { +public: + void armThrowAfterRunning() { _throwAfterRunning = true; } + + void post(std::function task) override { + ++_depth; + { + // The task runs as one of this executor's, as a real queue's would. + ::core::async::ExecutorScope const scope{coreExecutor()}; + task(); + } + --_depth; + if (_throwAfterRunning && _depth == 0) { + throw std::runtime_error{"post failed after running the task"}; + } + } + +private: + bool _throwAfterRunning = false; + int _depth = 0; +}; + +} // namespace + +TEST_CASE("co_await whose attach settles the await before its post fails resumes with the value, once", + "[coroutine][client]") { + // The attach is posted to the completion's executor, which runs it inline: + // the completion is already settled, so the attached handler settles the + // await and queues the resumption on `exec` -- and only then does the post + // throw. The await is already answered, so the throw must not reach the + // coroutine as well. + RunsThenThrowsExecutor executor; + auto [completion, promise] = Completion::makeSettleable(&executor); + promise.resolve(5); + executor.armThrowAfterRunning(); + morph::exec::MainThreadExecutor exec; + auto seen = std::make_shared(); + + morph::async::spawn(exec, awaitInto(std::move(completion), seen)); + + REQUIRE(pumpUntil(exec, [&] { return seen->finished.load(); })); + CHECK(seen->value == 5); + CHECK(seen->error.empty()); + CHECK_FALSE(seen->cancelled); +} diff --git a/tests/test_executor_extra.cpp b/tests/test_executor_extra.cpp index c778d0418..cad49bf14 100644 --- a/tests/test_executor_extra.cpp +++ b/tests/test_executor_extra.cpp @@ -31,9 +31,10 @@ TEST_CASE("morph::exec::MainThreadExecutor: tasks posted before runFor are all d REQUIRE(count.load() == numTasks); } -TEST_CASE("morph::exec::ThreadPoolExecutor: stop with queued tasks does not deadlock", "[executor]") { - // Post many tasks and immediately destroy the pool — destructor must join cleanly. - // Some tasks may not run; the important invariant is no deadlock/crash. +TEST_CASE("morph::exec::ThreadPoolExecutor: destruction runs every task queued before it, then joins", "[executor]") { + // Post many tasks and immediately destroy the pool: the workers drain the + // queue before they exit, so the join neither deadlocks nor drops a task + // that was queued in time. std::atomic ran{0}; { morph::exec::ThreadPoolExecutor pool{1}; @@ -43,10 +44,8 @@ TEST_CASE("morph::exec::ThreadPoolExecutor: stop with queued tasks does not dead ran.fetch_add(1); }); } - // pool destructor joins here — workers drain remaining tasks before exit } - // After join, whatever ran is fine; the test verifies it did not hang. - REQUIRE(ran.load() >= 0); + REQUIRE(ran.load() == 100); } TEST_CASE("morph::exec::ThreadPoolExecutor: exception in one task does not kill worker", "[executor]") { diff --git a/tests/test_file_io_ops.cpp b/tests/test_file_io_ops.cpp index f01c12c7f..0b7d7b7cc 100644 --- a/tests/test_file_io_ops.cpp +++ b/tests/test_file_io_ops.cpp @@ -13,13 +13,27 @@ #include #include #include +#include #include #include +#include #include +#include #include +#include namespace { +std::string readAll(const std::filesystem::path& path) { + std::ifstream in{path, std::ios::binary}; + return {std::istreambuf_iterator{in}, std::istreambuf_iterator{}}; +} + +void writeAll(const std::filesystem::path& path, std::string_view bytes) { + std::ofstream out{path, std::ios::binary}; + out << bytes; +} + std::filesystem::path tempIoPath(const std::string& tag) { static int counter = 0; return std::filesystem::temp_directory_path() / @@ -342,3 +356,107 @@ TEST_CASE("morph::core::repairTornTail: a read error leaves the file untouched", std::filesystem::remove_all(dir); } #endif + +TEST_CASE("morph::core::rollBackShortWrite: a short first write into an empty file is trimmed to nothing", + "[file_io_ops]") { + // Offset zero is a real offset -- the start of an empty file -- not the + // "could not query" sentinel, which is negative. + auto const path = tempIoPath("offset_zero"); + writeAll(path, ""); + + morph::core::FileIoOps ioOps; + { + auto const file = openForAppend(path); + REQUIRE(file != nullptr); + morph::core::positionAtEnd(file.get()); + long long const offsetBeforeWrite = morph::core::wideFtell(file.get()); + REQUIRE(offsetBeforeWrite == 0); + REQUIRE(std::fwrite("{\"tw", 1, 4, file.get()) == 4); + + CHECK(morph::core::rollBackShortWrite(ioOps, file.get(), path, offsetBeforeWrite) == + morph::core::RollBack::clean); + } + + CHECK(std::filesystem::file_size(path) == 0); + std::filesystem::remove(path); +} + +TEST_CASE("morph::core::rollBackShortWrite: a second short write on the same handle rolls back too", "[file_io_ops]") { + // The caller takes each rollback's offset from `wideFtell` just before the + // write, so after one rollback the handle's position has to describe the + // shortened file. A stale position would hand the second rollback an + // offset past the second partial record, and leave it on disk. + auto const path = tempIoPath("second_rollback"); + writeAll(path, "{\"one\":1}\n"); + + morph::core::FileIoOps ioOps; + { + auto const file = openForAppend(path); + REQUIRE(file != nullptr); + morph::core::positionAtEnd(file.get()); + for (int attempt = 0; attempt < 2; ++attempt) { + long long const offsetBeforeWrite = morph::core::wideFtell(file.get()); + CHECK(offsetBeforeWrite == 10); + REQUIRE(std::fwrite("{\"tw", 1, 4, file.get()) == 4); + CHECK(morph::core::rollBackShortWrite(ioOps, file.get(), path, offsetBeforeWrite) == + morph::core::RollBack::clean); + } + } + + CHECK(readAll(path) == "{\"one\":1}\n"); + std::filesystem::remove(path); +} + +// ── repairTornTail ────────────────────────────────────────────────────────── + +namespace { + +struct WarningCapture { + std::vector warnings; + morph::log::ScopedLoggerOverride guard{[this](morph::log::LogLevel level, std::string_view msg) { + if (level == morph::log::LogLevel::warn) { + warnings.emplace_back(msg); + } + }}; +}; + +} // namespace + +TEST_CASE("morph::core::repairTornTail: a file holding only a torn record is emptied", "[file_io_ops]") { + // No newline anywhere: there is no complete record, so everything goes. + auto const path = tempIoPath("only_torn"); + writeAll(path, "{\"tw"); + + morph::core::FileIoOps ioOps; + morph::core::repairTornTail(ioOps, path, "TestComponent"); + + CHECK(std::filesystem::file_size(path) == 0); + std::filesystem::remove(path); +} + +TEST_CASE("morph::core::repairTornTail: an intact file is left alone and reported as nothing", "[file_io_ops]") { + auto const path = tempIoPath("intact"); + writeAll(path, "{\"one\":1}\n{\"two\":2}\n"); + + WarningCapture const capture; + morph::core::FileIoOps ioOps; + morph::core::repairTornTail(ioOps, path, "TestComponent"); + + CHECK(readAll(path) == "{\"one\":1}\n{\"two\":2}\n"); + CHECK(capture.warnings.empty()); + std::filesystem::remove(path); +} + +TEST_CASE("morph::core::repairTornTail: a trimmed tail is reported with the bytes it discarded", "[file_io_ops]") { + auto const path = tempIoPath("trim_reported"); + writeAll(path, "{\"one\":1}\n{\"tw"); + + WarningCapture capture; + morph::core::FileIoOps ioOps; + morph::core::repairTornTail(ioOps, path, "TestComponent"); + + CHECK(readAll(path) == "{\"one\":1}\n"); + REQUIRE(capture.warnings.size() == 1); + CHECK(capture.warnings[0].starts_with("TestComponent: discarded 4 byte(s) of a torn trailing record in ")); + std::filesystem::remove(path); +} diff --git a/tests/test_graceful_shutdown.cpp b/tests/test_graceful_shutdown.cpp index 084fb2988..b82990588 100644 --- a/tests/test_graceful_shutdown.cpp +++ b/tests/test_graceful_shutdown.cpp @@ -238,6 +238,52 @@ TEST_CASE("RemoteServer::drainedWithin with no time to wait answers false while REQUIRE(execReply.await(2s)); } +TEST_CASE( + "RemoteServer::drainedWithin with no time to wait answers from the count its request finds, " + "not from a later one", + "[shutdown][graceful][drain]") { + gGSSlowStarted.store(0, std::memory_order_relaxed); + morph::exec::ThreadPoolExecutor pool{4}; + auto const server = std::make_shared(pool); + + morph::testing::WaitReply regReply; + server->handle(morph::wire::encode(morph::wire::makeRegister("GS_SlowModel")), std::ref(regReply)); + REQUIRE(regReply.await()); + + morph::wire::Envelope req; + req.kind = "execute"; + req.callId = 1; + req.modelId = regReply.env.modelId; + req.modelType = "GS_SlowModel"; + req.actionType = "GS_SlowAction"; + req.body = R"({"ms":50})"; + morph::testing::WaitReply execReply; + server->handle(morph::wire::encode(req), std::ref(execReply)); + REQUIRE(morph::testing::waitUntil([] { return gGSSlowStarted.load(std::memory_order_relaxed) >= 1; })); + + // Hold the server strand, so the drain request and the execute's + // completion queue on it in that order: the request is answered while the + // execute is still counted, and the completion right behind it must not + // turn that answer into `true`. + std::atomic release{false}; + std::atomic holding{false}; + server->strand().post([&release, &holding] { + holding.store(true); + while (!release.load()) { + std::this_thread::sleep_for(1ms); + } + }); + REQUIRE(morph::testing::waitUntil([&holding] { return holding.load(); })); + + morph::exec::MainThreadExecutor owner; + auto drained = server->drainedWithin(0ms, owner); + REQUIRE(execReply.await(2s)); + release.store(true); + + bool const answer = morph::testing::awaitValueOn(owner, std::move(drained)); + CHECK_FALSE(answer); +} + TEST_CASE("RemoteServer::drainedWithin times out while a slower-than-deadline execute is still running", "[shutdown][graceful][drain]") { gGSSlowStarted.store(0, std::memory_order_relaxed); diff --git a/tests/test_journal_payload_evolution.cpp b/tests/test_journal_payload_evolution.cpp index 852a58af3..3069dae17 100644 --- a/tests/test_journal_payload_evolution.cpp +++ b/tests/test_journal_payload_evolution.cpp @@ -285,6 +285,68 @@ TEST_CASE("payloadShapeString: the deepest level that still renders is kPayloadS REQUIRE(payloadFingerprint() != payloadFingerprint()); } +// Every other nesting form costs one level per wrapper too, so nine of any of +// them put the leaf past the limit. + +/// One reflected member wrapped around @p T. +template +struct PEBox { + T v; +}; + +/// A custom-codec wrapper around @p T that declares its inner shape. +template +struct PEWrap { + T inner; +}; + +template +struct morph::model::PayloadShapeTag> { + using Inner = T; + static constexpr std::string_view name() { return "w"; } +}; + +/// @p Wrap applied @p N times around `std::int32_t`. +template