diff --git a/cpp/implementations/plain/plain_logos_object.cpp b/cpp/implementations/plain/plain_logos_object.cpp index 5fd08a0..130da13 100644 --- a/cpp/implementations/plain/plain_logos_object.cpp +++ b/cpp/implementations/plain/plain_logos_object.cpp @@ -9,13 +9,99 @@ #include #include +#include +#include #include #include +#include #include #include namespace logos::plain { +namespace { + +// How long a waiter sleeps before it looks at the stop flag again. +// +// A std::future wait cannot be interrupted, so the only way to make one +// abandonable is to wait in slices against the same overall deadline and check +// the flag between them. That slice IS the teardown-latency bound: releasing a +// handle with a call in flight costs at most one of these, instead of whatever +// is left of the call's timeout (20s on the protocol default, logos_mode.h). +// +// 25ms is chosen off both ends of the trade: +// * latency — a module unloading mid-call should feel instantaneous. 25ms is +// under two 60Hz frames and well below the ~100ms at which a stall becomes +// perceptible, so even a shutdown releasing handles back to back stays +// invisible. +// * cost — one timed wakeup per slice per IN-FLIGHT call: 40/s, i.e. 800 +// spread over a full 20s default timeout, and paid only while a call is +// actually outstanding. That is far below the wakeup rate of the Qt event +// loop these waiters already sit beside. +// Below ~5ms the extra wakeups buy latency nobody can perceive; at 100-250ms +// the teardown hitch starts to show. +constexpr std::chrono::milliseconds kWaitSlice(25); + +enum class WaitOutcome { Ready, TimedOut, Cancelled }; + +// The interruptible form of `fut.wait_for(milliseconds(timeoutMs))`. +// +// The overall deadline is computed once, so slicing does not stretch the +// timeout the caller asked for: the last slice ends exactly on it. +WaitOutcome waitForResult(std::future& fut, int timeoutMs, + const std::atomic& stopping) +{ + using clock = std::chrono::steady_clock; + const auto deadline = clock::now() + std::chrono::milliseconds(timeoutMs); + for (;;) { + // Checked BEFORE sleeping, so a stop that already happened costs + // nothing, and after every slice, so one that arrives mid-wait costs at + // most kWaitSlice. A pending stop beats a result that landed in the + // same tick on purpose: the caller has released the handle and is no + // longer interested in the answer. + if (stopping.load(std::memory_order_acquire)) + return WaitOutcome::Cancelled; + + const auto remaining = deadline - clock::now(); + const bool expired = remaining <= clock::duration::zero(); + // At (or past) the deadline, poll once with a zero wait rather than + // giving up blind — that is what the single unsliced wait_for did for a + // non-positive timeout, and a future that is already ready must still + // be reported as ready. + const auto slice = expired + ? clock::duration::zero() + : std::min(kWaitSlice, remaining); + if (fut.wait_for(slice) == std::future_status::ready) + return WaitOutcome::Ready; + if (expired) + return WaitOutcome::TimedOut; + } +} + +// The honest code for "the object was released while your call was in flight". +// +// logos_call_error.h's vocabulary is part of the wire contract, so this reuses +// it rather than minting a code. "transport_error" is defined there as "the +// connection failed or was torn down mid-call", which is exactly what happened: +// the consumer tore its own end of the call channel down. Every alternative in +// that set misattributes the failure — "object_unavailable" says the module is +// not there (it is, and it is very likely about to answer; callers re-acquire +// on that code), "call_failed" blames the peer for a dispatch it performed +// perfectly well, and "timeout" — what this used to report, after waiting the +// deadline out — claims a deadline elapsed that did not. It is also already the +// code the wire produces for the same event seen from the other end: +// callErrorFromWire maps TRANSPORT_CLOSED / TRANSPORT_ERROR to transport_error. +logos::CallError callErrorReleased(const std::string& objectName, + const std::string& method) +{ + return logos::callErrorTransport( + objectName, + "call to '" + objectName + "." + method + "' was abandoned: the object " + "was released while the call was in flight"); +} + +} // anonymous namespace + PlainLogosObject::PlainLogosObject(std::string objectName, std::shared_ptr conn) : m_objectName(std::move(objectName)) @@ -26,16 +112,35 @@ PlainLogosObject::PlainLogosObject(std::string objectName, PlainLogosObject::~PlainLogosObject() { disconnectEvents(); - joinWaiters(); + stopAndJoinWaiters(); } -void PlainLogosObject::joinWaiters() +void PlainLogosObject::stopWaiters() { + { + // Published under m_completionMu — the mutex awaitCompletion evaluates + // its predicate under — so a waiter cannot read `false`, decide to + // sleep, and only then miss the notify_all below. The sliced future + // wait reads the same flag lock-free, which is why it is an atomic + // rather than a plain bool guarded by this mutex. + std::lock_guard g(m_completionMu); + m_stopping.store(true, std::memory_order_release); + } + m_completionCv.notify_all(); +} + +void PlainLogosObject::stopAndJoinWaiters() +{ + stopWaiters(); + std::vector waiters; { std::lock_guard g(m_waiterMu); waiters.swap(m_waiters); } + // Joined with NO lock held: a waiter on its way out still takes + // m_completionMu (awaitCompletion) and m_waiterMu is what a concurrent + // callMethodAsyncWithError needs to see the stop flag. for (auto& t : waiters) { if (t.joinable()) t.join(); @@ -139,18 +244,34 @@ QVariant PlainLogosObject::awaitCompletion(const QString& callId, int timeoutMs, const auto effectiveMs = timeoutMs > 0 ? timeoutMs : 30000; const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(effectiveMs); - const bool got = m_completionCv.wait_until(lk, deadline, - [&] { return m_completions.count(callId) > 0; }); - if (!got) { - qWarning() << "PlainLogosObject: deferred call" << callId << "timed out"; + // Unlike the future wait this one is interruptible by construction: widen + // the predicate, and stopWaiters()' notify_all does the rest. No slicing, so + // no latency floor at all here — a stop wakes this wait immediately. + m_completionCv.wait_until(lk, deadline, [&] { + return m_completions.count(callId) > 0 + || m_stopping.load(std::memory_order_relaxed); + }); + + // A completion that actually landed beats a concurrent stop: there is a real + // answer in hand, so hand it over rather than manufacture an error. + const auto it = m_completions.find(callId); + if (it != m_completions.end()) { + const QVariant result = it->second; + m_completions.erase(it); + return result; + } + if (m_stopping.load(std::memory_order_relaxed)) { + qWarning() << "PlainLogosObject: deferred call" << callId + << "abandoned — object released while it was in flight"; if (err) - *err = logos::callErrorTimeout(m_objectName, methodName.toStdString(), - effectiveMs); + *err = callErrorReleased(m_objectName, methodName.toStdString()); return QVariant(); } - const QVariant result = m_completions[callId]; - m_completions.erase(callId); - return result; + qWarning() << "PlainLogosObject: deferred call" << callId << "timed out"; + if (err) + *err = logos::callErrorTimeout(m_objectName, methodName.toStdString(), + effectiveMs); + return QVariant(); } namespace { @@ -165,6 +286,14 @@ namespace { // process, regardless of which worker thread completed the future. // If the application has shut down (instance() is null), we drop the // callback rather than invoke it from an arbitrary thread. +// +// Deliberately a FREE function taking everything BY VALUE, and deliberately not +// a member: the queued lambda runs on a later event-loop iteration, which for a +// waiter cancelled by teardown is after the PlainLogosObject is already gone. +// Nothing it touches may belong to the object — which is why the waiter copies +// objectName/method up front instead of reading m_objectName from inside here. +// Do not give this a `this`; delivering during teardown would become the +// use-after-free that joining the waiters exists to prevent. void postToQtEventLoop(PlainLogosObject::AsyncResultErrorCallback callback, QVariant result, logos::CallError err) { @@ -230,8 +359,8 @@ void PlainLogosObject::callMethodAsyncWithError(const QString& authToken, // io_context (the connection already runs on it) so we don't spin // up a thread per pending RPC. // - // The thread is JOINed in joinWaiters() (destructor / release), not - // detached: capturing `this` for awaitCompletion / m_objectName is + // The thread is JOINed in stopAndJoinWaiters() (destructor / release), not + // detached: capturing `this` for awaitCompletion / m_stopping is // only safe while the object is alive, and release() used to // `delete this` while a waiter could still be mid-flight. const std::string objectName = m_objectName; @@ -240,10 +369,36 @@ void PlainLogosObject::callMethodAsyncWithError(const QString& authToken, // detach-then-push left a window where delete this raced the waiter. { std::lock_guard g(m_waiterMu); + if (m_stopping.load(std::memory_order_acquire)) { + // Teardown has already swapped m_waiters out, so a thread pushed + // now would never be joined — exactly the dangling waiter this + // whole mechanism exists to prevent. Answer as a cancelled call + // instead, which keeps the exactly-once contract either way. + postToQtEventLoop(std::move(callback), QVariant(), + callErrorReleased(objectName, method)); + return; + } m_waiters.emplace_back([this, objectName, fut, timeoutMs, methodName, method, callback = std::move(callback)]() mutable { - if (fut->wait_for(std::chrono::milliseconds(timeoutMs)) - != std::future_status::ready) { + // Everything reached through `this` below (m_stopping, + // awaitCompletion's m_completionMu / m_completions) is safe only + // because stopAndJoinWaiters() joins this thread before the object + // dies. Everything handed to postToQtEventLoop is a COPY, because + // that delivery happens after this thread has returned — i.e. + // possibly after the object is gone. Keep it that way. + const WaitOutcome outcome = waitForResult(*fut, timeoutMs, m_stopping); + if (outcome == WaitOutcome::Cancelled) { + // A cancelled call still DELIVERS, exactly once. Returning + // silently here would honour the "stop fast" half and break the + // half that matters more: callMethodAsyncWithError (and + // lp_invoke_async above it) promise the callback fires exactly + // once, so a dropped one turns a bounded stall into an + // unbounded hang in every caller that awaits it. + postToQtEventLoop(std::move(callback), QVariant(), + callErrorReleased(objectName, method)); + return; + } + if (outcome == WaitOutcome::TimedOut) { postToQtEventLoop(std::move(callback), QVariant(), logos::callErrorTimeout(objectName, method, timeoutMs)); @@ -258,7 +413,10 @@ void PlainLogosObject::callMethodAsyncWithError(const QString& authToken, } QVariant value = rpcValueToQVariant(res.value); // Resolve a "multi" provider's deferred completion (sentinel → wait for - // the completion event) right here on the waiter thread. + // the completion event) right here on the waiter thread. This is the + // second interruptible site: a stop lands it on callErrorReleased, + // which still falls through to the single post below — one callback, + // whichever way this went. logos::CallError err; { QString callId; @@ -355,11 +513,14 @@ void PlainLogosObject::release() // own events and drop our reference — the connection stays alive // until PlainTransportConnection itself is destroyed. // - // joinWaiters() before delete: in-flight async waiters capture `this` + // stopAndJoinWaiters() before delete: in-flight async waiters capture `this` // (for awaitCompletion). Detaching them used to let release() free the - // object under a still-running waiter. + // object under a still-running waiter — and merely joining them made + // release() block for the rest of the call's timeout, so they are asked to + // stop first. Each abandoned call still delivers its callback, once, with + // callErrorReleased. disconnectEvents(); - joinWaiters(); + stopAndJoinWaiters(); m_conn.reset(); delete this; } diff --git a/cpp/implementations/plain/plain_logos_object.h b/cpp/implementations/plain/plain_logos_object.h index 105e2ae..122ecb3 100644 --- a/cpp/implementations/plain/plain_logos_object.h +++ b/cpp/implementations/plain/plain_logos_object.h @@ -5,6 +5,7 @@ #include "rpc_connection.h" +#include #include #include #include @@ -77,17 +78,29 @@ private: // matching callId lands. The completion arrives on the connection's IO // thread; the caller waits on another thread — m_completionMu/Cv bridge them. void ensureCompletionSub(); - // `err` (optional) receives the timeout when the completion never lands — - // a deferred call that gives up is a timeout like any other, and used to be - // reported as a null result. + // `err` (optional) receives the reason when no completion lands: the + // timeout when the deadline elapses (a deferred call that gives up is a + // timeout like any other, and used to be reported as a null result), or a + // transport error when the object is released out from under the wait. QVariant awaitCompletion(const QString& callId, int timeoutMs, const QString& methodName = QString(), logos::CallError* err = nullptr); - // Join every per-call waiter before tearing the object down. callMethodAsync - // used to detach those threads, so release()/delete racing an in-flight - // wait was a use-after-free on `this` (m_objectName, awaitCompletion, …). - void joinWaiters(); + // Ask every in-flight waiter to give up, then join them, then return. + // + // The JOIN is what makes the waiters safe at all: they capture `this` (they + // read m_stopping and call awaitCompletion), and callMethodAsync used to + // DETACH them, so release()/delete racing an in-flight wait was a + // use-after-free. But joining alone means teardown blocks for whatever is + // left of the call's timeout — up to 20s on the protocol default — because + // a waiter has no reason to return early. Hence the stop first: it costs + // one wait slice instead, and a cancelled call still delivers its callback + // exactly once (with an error), because dropping it would turn the stall + // into a permanent hang in the caller awaiting it. + void stopAndJoinWaiters(); + // Raise the stop flag and wake anything parked on m_completionCv. Split out + // because the flag has to be published under m_completionMu (see the .cpp). + void stopWaiters(); std::string m_objectName; std::shared_ptr m_conn; @@ -101,6 +114,11 @@ private: std::mutex m_waiterMu; std::vector m_waiters; + // Read lock-free by the sliced future wait and under m_completionMu by + // awaitCompletion's predicate; written under m_completionMu so the + // condition-variable side cannot miss it. Never cleared — an object that + // has begun tearing down does not come back. + std::atomic m_stopping{false}; }; } // namespace logos::plain diff --git a/tests/protocol/CMakeLists.txt b/tests/protocol/CMakeLists.txt index fcc50ed..0b03a42 100644 --- a/tests/protocol/CMakeLists.txt +++ b/tests/protocol/CMakeLists.txt @@ -50,6 +50,15 @@ add_executable(protocol_tests # is not silently plain-TCP-only. Exercises the deferred-completion timeout, # which needs no blocking provider and therefore no second event loop. test_call_error_qt_remote.cpp + # Teardown of a PlainLogosObject with a call still IN FLIGHT. Joining the + # per-call waiters closed a use-after-free but left release() blocking for + # the remainder of the call's timeout (up to the 20s default), because + # joinWaiters() could only join, never ask a waiter to stop. Pins all three + # halves of the fix: teardown costs one wait slice, the callback still fires + # EXACTLY ONCE on every outcome including cancellation (a waiter that just + # returns turns a stall into a hang), and the join that keeps `this` alive + # under the waiter is still there. + test_plain_object_teardown.cpp # Component tests that moved here with their code (from logos-cpp-sdk) test_token_manager.cpp test_mock_store.cpp diff --git a/tests/protocol/test_plain_object_teardown.cpp b/tests/protocol/test_plain_object_teardown.cpp new file mode 100644 index 0000000..3e7ac7c --- /dev/null +++ b/tests/protocol/test_plain_object_teardown.cpp @@ -0,0 +1,572 @@ +// Tearing down a PlainLogosObject with a call still in flight. +// +// The branch this sits on stopped callMethodAsync from DETACHING its waiter +// thread: the waiter captures `this` (it reads m_objectName and calls +// awaitCompletion), and release() used to `delete this` underneath it. Waiters +// are now registered in m_waiters and joined before the object dies. +// +// Joining alone only trades one bug for a stall. The join-only joinWaiters() +// had no way to ASK a waiter to stop, so both of the waiter's blocking sites +// ran to their deadline: +// +// * the std::future wait in callMethodAsyncWithError, and +// * the completion-event wait in awaitCompletion (a "multi" provider's +// deferred result). +// +// Destroying a handle with an in-flight call therefore blocked for the +// remainder of that call's timeout — up to 20s on the protocol default +// (logos_mode.h Timeout). A module unloading mid-call stalls the unload for +// that long, on whichever thread called release(). +// +// What these tests pin, and why each one is here: +// +// 1. TEARDOWN LATENCY. release() with an in-flight call must return in about +// one wait slice, not one call timeout. +// +// 2. THE CALLBACK CONTRACT, which is the part a naive fix breaks. +// callMethodAsyncWithError (and lp_invoke_async above it) promise the +// callback fires EXACTLY ONCE. A waiter that simply RETURNS when asked to +// stop silently drops it — trading a bounded stall for an unbounded hang +// in any caller that awaits that callback. So the three outcomes are +// counted, not just observed: normal completion, timeout, and +// cancellation-by-teardown must each deliver exactly one callback. Zero +// and two are both failures. +// +// 3. THE UAF THAT MUST NOT COME BACK. Cancellation must not become "let the +// waiter go"; the join still has to happen. The release-during-call race +// is hammered here so an ASan/TSan build has something to catch. +// +// Everything runs against a live in-process PlainTransportHost over real TCP, +// and drives PlainLogosObject directly (PlainTransportConnection::requestObject) +// so release() is measured on its own rather than through lp_client_destroy. + +#include + +#include "logos_async_dispatch.h" +#include "logos_call_error.h" +#include "logos_object.h" +#include "logos_provider_interface.h" +#include "logos_transport_config.h" +#include "module_proxy.h" + +#include "plain_transport_connection.h" +#include "plain_transport_host.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace logos::plain; + +namespace { + +// A provider with a method that parks until the test lets it go. The existing +// suites use a fixed sleep, which cannot express "in flight for as long as the +// test needs": a sleep that is shorter than the call timeout makes the future +// ready on its own and the teardown measurement then times the sleep instead of +// the timeout. +class BlockingProvider : public LogosProviderObject { +public: + QVariant callMethod(const QString& method, const QVariantList& args) override + { + if (method == QLatin1String("ping")) return args.value(0, QVariant(1)); + if (method == QLatin1String("block")) { + std::unique_lock lk(m_mu); + m_cv.wait(lk, [this] { return m_released; }); + return QVariant(42); + } + // The other in-flight shape: a "multi" provider that answers the + // pending sentinel straight away and then never pushes the completion + // event, so the consumer parks in awaitCompletion instead of on the + // future. Returns immediately, so unlike `block` it holds no thread. + if (method == QLatin1String("defer")) { + QVariantMap sentinel; + sentinel[logos::pendingCallKey()] = QStringLiteral("never-completes"); + return sentinel; + } + return QVariant(); + } + + void letGo() + { + { + std::lock_guard g(m_mu); + m_released = true; + } + m_cv.notify_all(); + } + + QJsonArray getMethods() override { return QJsonArray{}; } + bool informModuleToken(const QString&, const QString&) override { return true; } + void setEventListener(EventCallback) override {} + void init(void*) override {} + QString providerName() const override { return QStringLiteral("blocker"); } + QString providerVersion() const override { return QStringLiteral("1.0.0"); } + +private: + std::mutex m_mu; + std::condition_variable m_cv; + bool m_released = false; +}; + +QCoreApplication* ensureApp() +{ + static int argc = 0; + static char* argv[] = { nullptr }; + if (!QCoreApplication::instance()) + new QCoreApplication(argc, argv); + return QCoreApplication::instance(); +} + +// A live host publishing `blocker_module` through a ModuleProxy on its own +// thread — the provider blocks, so it must not be the thread the consumer needs +// to deliver its callbacks on. +class LiveHost { +public: + LiveHost() + { + LogosTransportConfig cfg; + cfg.protocol = LogosProtocol::Tcp; + cfg.host = "127.0.0.1"; + cfg.port = 0; // ephemeral + m_host = std::make_unique(cfg); + m_started = m_host->start(); + + m_proxy = new ModuleProxy(&m_provider); + m_proxy->saveToken(QStringLiteral("origin"), QStringLiteral("live-token")); + m_thread = new QThread; + m_proxy->moveToThread(m_thread); + m_thread->start(); + + m_published = m_host->publishObject("blocker_module", m_proxy); + + const QString endpoint = m_host->endpoint(); + m_port = endpoint.mid(endpoint.lastIndexOf(':') + 1).toUShort(); + } + + ~LiveHost() + { + // Anything still parked in the provider would deadlock the thread quit + // below; let every blocked (and queued) call finish first. + m_provider.letGo(); + QCoreApplication::processEvents(QEventLoop::AllEvents, 200); + m_host.reset(); + QCoreApplication::processEvents(QEventLoop::AllEvents, 50); + m_thread->quit(); + m_thread->wait(); + delete m_proxy; + delete m_thread; + } + + bool ok() const { return m_started && m_published && m_port != 0; } + uint16_t port() const { return m_port; } + BlockingProvider& provider() { return m_provider; } + +private: + BlockingProvider m_provider; + std::unique_ptr m_host; + ModuleProxy* m_proxy = nullptr; + QThread* m_thread = nullptr; + bool m_started = false; + bool m_published = false; + uint16_t m_port = 0; +}; + +// A consumer-side connection to that host. requestObject() hands back a bare +// PlainLogosObject, so release() is exercised directly. +std::unique_ptr connectTo(uint16_t port) +{ + LogosTransportConfig cfg; + cfg.protocol = LogosProtocol::Tcp; + cfg.host = "127.0.0.1"; + cfg.port = port; + auto conn = std::make_unique(cfg); + if (!conn->connectToHost()) return nullptr; + return conn; +} + +// Counts callbacks. `count` is the assertion that matters: the contract is +// EXACTLY ONE, so both 0 and 2 must fail. +struct Sink { + std::atomic count{0}; + std::mutex mu; + QVariant value; + logos::CallError err; + + std::string code() + { + std::lock_guard g(mu); + return err.code; + } + std::string message() + { + std::lock_guard g(mu); + return err.message; + } +}; + +// The callback CO-OWNS its sink. A test that fails its "the callback fired" +// assertion returns with the delivery still queued on the Qt event loop, and a +// sink captured by reference would be a dead stack frame by then — a genuine +// regression would surface as a crash in the harness instead of the clean +// assertion failure that names it. +LogosObjectErrorChannel::AsyncResultErrorCallback cbFor(std::shared_ptr sink) +{ + return [sink](QVariant v, const logos::CallError& e) { + std::lock_guard g(sink->mu); + sink->value = std::move(v); + sink->err = e; + sink->count.fetch_add(1); + }; +} + +void pump(int ms) +{ + QElapsedTimer t; + t.start(); + while (t.elapsed() < ms) + QCoreApplication::processEvents(QEventLoop::AllEvents, 10); +} + +bool pumpUntilFired(Sink& s, int budgetMs) +{ + QElapsedTimer t; + t.start(); + while (s.count.load() == 0 && t.elapsed() < budgetMs) + QCoreApplication::processEvents(QEventLoop::AllEvents, 10); + return s.count.load() > 0; +} + +LogosObjectErrorChannel* channelFor(LogosObject* obj) +{ + return dynamic_cast(obj); +} + +const char* kToken = "live-token"; + +// Long enough that a full-timeout teardown is unmistakable next to a +// one-slice one, short enough that a regression doesn't wedge CI for 20s. +constexpr int kLongTimeoutMs = 8000; + +// The budget release() must fit in. One wait slice is 25ms; this leaves an +// order of magnitude of headroom for a loaded CI box while still being ~10x +// below the call timeout above. +constexpr int kTeardownBudgetMs = 750; + +} // namespace + +class PlainObjectTeardownTest : public ::testing::Test { +protected: + void SetUp() override { ensureApp(); } +}; + +// ── 1. teardown latency ───────────────────────────────────────────────────── +// +// The provider parks forever, the call is given 8s, and then the handle is +// released. Pre-fix release() sat inside joinWaiters() until the waiter's own +// future wait hit 8000ms, because nothing could tell it to stop. +TEST_F(PlainObjectTeardownTest, ReleaseWithACallInFlightDoesNotWaitOutTheTimeout) +{ + LiveHost host; + ASSERT_TRUE(host.ok()); + auto conn = connectTo(host.port()); + ASSERT_NE(conn, nullptr); + + LogosObject* obj = conn->requestObject(QStringLiteral("blocker_module"), 5000); + ASSERT_NE(obj, nullptr); + auto* ch = channelFor(obj); + ASSERT_NE(ch, nullptr); + + auto sink = std::make_shared(); + ch->callMethodAsyncWithError(kToken, QStringLiteral("block"), {}, + kLongTimeoutMs, cbFor(sink)); + // Let the call reach the provider and park there, so the waiter really is + // mid-wait when release() lands. + pump(200); + EXPECT_EQ(sink->count.load(), 0) << "the provider answered; nothing was in flight"; + + QElapsedTimer timer; + timer.start(); + obj->release(); + const qint64 releaseMs = timer.elapsed(); + + std::cout << " release() with an in-flight " << kLongTimeoutMs + << "ms call took " << releaseMs << "ms" << std::endl; + + EXPECT_LT(releaseMs, kTeardownBudgetMs) + << "release() waited out the call timeout instead of cancelling the waiter"; + + pumpUntilFired(*sink, 2000); + host.provider().letGo(); + pump(200); +} + +// ── 2. the callback contract, all three outcomes ──────────────────────────── + +TEST_F(PlainObjectTeardownTest, NormalCompletionFiresTheCallbackExactlyOnce) +{ + LiveHost host; + ASSERT_TRUE(host.ok()); + auto conn = connectTo(host.port()); + ASSERT_NE(conn, nullptr); + + LogosObject* obj = conn->requestObject(QStringLiteral("blocker_module"), 5000); + ASSERT_NE(obj, nullptr); + auto* ch = channelFor(obj); + ASSERT_NE(ch, nullptr); + + auto sink = std::make_shared(); + ch->callMethodAsyncWithError(kToken, QStringLiteral("ping"), + QVariantList{ QVariant(7) }, 5000, cbFor(sink)); + ASSERT_TRUE(pumpUntilFired(*sink, 10000)) << "the callback never fired"; + // Keep pumping: a second delivery would arrive here. + pump(300); + + std::cout << " normal completion -> callbacks=" << sink->count.load() + << " code='" << sink->code() << "'" << std::endl; + + EXPECT_EQ(sink->count.load(), 1); + EXPECT_TRUE(sink->code().empty()) << "a successful call reported an error"; + { + std::lock_guard g(sink->mu); + EXPECT_EQ(sink->value.toInt(), 7); + } + + obj->release(); +} + +TEST_F(PlainObjectTeardownTest, TimeoutFiresTheCallbackExactlyOnce) +{ + LiveHost host; + ASSERT_TRUE(host.ok()); + auto conn = connectTo(host.port()); + ASSERT_NE(conn, nullptr); + + LogosObject* obj = conn->requestObject(QStringLiteral("blocker_module"), 5000); + ASSERT_NE(obj, nullptr); + auto* ch = channelFor(obj); + ASSERT_NE(ch, nullptr); + + auto sink = std::make_shared(); + ch->callMethodAsyncWithError(kToken, QStringLiteral("block"), {}, 400, cbFor(sink)); + ASSERT_TRUE(pumpUntilFired(*sink, 10000)) << "the callback never fired"; + pump(400); + + std::cout << " timeout -> callbacks=" << sink->count.load() + << " code='" << sink->code() << "'" << std::endl; + + EXPECT_EQ(sink->count.load(), 1); + EXPECT_EQ(sink->code(), "timeout") + << "slicing the wait must not change what a real timeout reports"; + + obj->release(); + host.provider().letGo(); + pump(200); +} + +// The one a "just return on stop" fix breaks: the call is abandoned, and the +// caller must still be told — once — and told the truth. +// +// "transport_error" is the honest code. logos_call_error.h defines it as "the +// connection failed or was torn down mid-call", which is exactly this: the +// consumer tore its own end of the call channel down while the call was in +// flight. The alternatives lie about who failed — "object_unavailable" means +// the module is not there (it is, and is very likely about to answer, and +// callers re-acquire on that code), and "call_failed" blames the peer for a +// dispatch it performed perfectly well. It is also what the wire already +// reports for the same event seen from the other side: callErrorFromWire maps +// TRANSPORT_CLOSED to transport_error. +TEST_F(PlainObjectTeardownTest, CancellationByTeardownFiresTheCallbackExactlyOnce) +{ + LiveHost host; + ASSERT_TRUE(host.ok()); + auto conn = connectTo(host.port()); + ASSERT_NE(conn, nullptr); + + LogosObject* obj = conn->requestObject(QStringLiteral("blocker_module"), 5000); + ASSERT_NE(obj, nullptr); + auto* ch = channelFor(obj); + ASSERT_NE(ch, nullptr); + + auto sink = std::make_shared(); + ch->callMethodAsyncWithError(kToken, QStringLiteral("block"), {}, + kLongTimeoutMs, cbFor(sink)); + pump(200); + ASSERT_EQ(sink->count.load(), 0); + + obj->release(); + + ASSERT_TRUE(pumpUntilFired(*sink, 2000)) + << "the cancelled call dropped its callback — the caller waits forever"; + pump(400); // a second delivery would land here + + std::cout << " cancelled by release -> callbacks=" << sink->count.load() + << " code='" << sink->code() << "' message='" << sink->message() + << "'" << std::endl; + + EXPECT_EQ(sink->count.load(), 1); + EXPECT_EQ(sink->code(), "transport_error"); + EXPECT_FALSE(sink->message().empty()); + + host.provider().letGo(); + pump(200); +} + +// ── the SECOND blocking site: the deferred-completion wait ────────────────── +// +// A waiter has two places it can be parked, and cancelling only the first would +// be half a fix. Once a "multi" provider answers the pending sentinel, the +// waiter leaves the future wait entirely and blocks in awaitCompletion on +// m_completionCv until the completion event lands or the deadline passes. The +// provider here answers the sentinel and never completes, so release() lands +// while the waiter is in that second wait — not the first. +// +// This one interrupts with no latency floor at all: it is a condition variable, +// so the stop wakes it immediately rather than at the next slice boundary. +TEST_F(PlainObjectTeardownTest, ReleaseDuringADeferredCompletionCancelsThatWaitToo) +{ + LiveHost host; + ASSERT_TRUE(host.ok()); + auto conn = connectTo(host.port()); + ASSERT_NE(conn, nullptr); + + LogosObject* obj = conn->requestObject(QStringLiteral("blocker_module"), 5000); + ASSERT_NE(obj, nullptr); + auto* ch = channelFor(obj); + ASSERT_NE(ch, nullptr); + + auto sink = std::make_shared(); + ch->callMethodAsyncWithError(kToken, QStringLiteral("defer"), {}, + kLongTimeoutMs, cbFor(sink)); + // The sentinel comes back fast; this is long enough for the waiter to have + // left the future wait and be sitting in awaitCompletion. + pump(300); + ASSERT_EQ(sink->count.load(), 0) << "the deferred call completed on its own"; + + QElapsedTimer timer; + timer.start(); + obj->release(); + const qint64 releaseMs = timer.elapsed(); + + ASSERT_TRUE(pumpUntilFired(*sink, 2000)) << "the deferred call dropped its callback"; + pump(300); + + std::cout << " cancelled mid-defer -> release=" << releaseMs + << "ms callbacks=" << sink->count.load() + << " code='" << sink->code() << "'" << std::endl; + + EXPECT_LT(releaseMs, kTeardownBudgetMs) + << "release() waited out the deferred-completion deadline"; + EXPECT_EQ(sink->count.load(), 1); + EXPECT_EQ(sink->code(), "transport_error"); +} + +// A call started on an already-released... there is no such thing (release +// deletes), but a handle CAN be torn down between the call being issued and the +// waiter starting. Same contract: one callback. +TEST_F(PlainObjectTeardownTest, ReleaseImmediatelyAfterTheCallStillDeliversOnce) +{ + LiveHost host; + ASSERT_TRUE(host.ok()); + auto conn = connectTo(host.port()); + ASSERT_NE(conn, nullptr); + + LogosObject* obj = conn->requestObject(QStringLiteral("blocker_module"), 5000); + ASSERT_NE(obj, nullptr); + auto* ch = channelFor(obj); + ASSERT_NE(ch, nullptr); + + auto sink = std::make_shared(); + ch->callMethodAsyncWithError(kToken, QStringLiteral("block"), {}, + kLongTimeoutMs, cbFor(sink)); + obj->release(); // no pump: the waiter may not even have started + + ASSERT_TRUE(pumpUntilFired(*sink, 2000)) << "callback dropped"; + pump(300); + + std::cout << " released instantly -> callbacks=" << sink->count.load() + << " code='" << sink->code() << "'" << std::endl; + + EXPECT_EQ(sink->count.load(), 1); + + host.provider().letGo(); + pump(200); +} + +// ── 3. the UAF must stay closed ───────────────────────────────────────────── +// +// The waiter must never outlive the object: it reads m_stopping and may call +// awaitCompletion (m_completionMu, m_completions) after the stop, so the join +// is what keeps `this` alive underneath it. Cancelling must not turn into +// detaching. +// +// Hammered with a varying gap between issuing the call and releasing, so the +// release lands at different points of the waiter's startup. Plain, this +// catches a dropped or doubled callback; run under a UAF detector it catches +// the freed `this` directly. Verified with macOS Guard Malloc +// (DYLD_INSERT_LIBRARIES=/usr/lib/libgmalloc.dylib): clean as written, SIGSEGV +// the moment the join is turned back into a detach. ASan/TSan are not usable +// on this toolchain — libclang_rt livelocks in its own init before main. +TEST_F(PlainObjectTeardownTest, ReleaseRacingTheWaiterIsSafeAndDeliversOnce) +{ + LiveHost host; + ASSERT_TRUE(host.ok()); + auto conn = connectTo(host.port()); + ASSERT_NE(conn, nullptr); + + constexpr int kRounds = 60; + int delivered = 0; + QElapsedTimer total; + total.start(); + + for (int i = 0; i < kRounds; ++i) { + LogosObject* obj = conn->requestObject(QStringLiteral("blocker_module"), 5000); + ASSERT_NE(obj, nullptr); + auto* ch = channelFor(obj); + ASSERT_NE(ch, nullptr); + + auto sink = std::make_shared(); + ch->callMethodAsyncWithError(kToken, QStringLiteral("block"), {}, + kLongTimeoutMs, cbFor(sink)); + // 0..~1.5ms of drift across the rounds, sweeping the window between + // registering the waiter and the waiter reaching its first wait. + if (i % 3 != 0) + QThread::usleep(static_cast((i % 30) * 50)); + + obj->release(); + + ASSERT_TRUE(pumpUntilFired(*sink, 3000)) << "round " << i << ": callback dropped"; + pump(20); + ASSERT_EQ(sink->count.load(), 1) << "round " << i << ": callback fired twice"; + ++delivered; + } + + const qint64 elapsed = total.elapsed(); + std::cout << " " << delivered << "/" << kRounds + << " release-during-call rounds delivered exactly once in " + << elapsed << "ms" << std::endl; + EXPECT_EQ(delivered, kRounds); + // 60 rounds x kLongTimeoutMs is 8 minutes if the stop stops working, which + // would otherwise show up only as a suite that got mysteriously slower. + // Post-fix a round costs a slice plus a round trip (~40ms), so this is an + // order of magnitude of headroom. + EXPECT_LT(elapsed, 30000) + << "rounds are waiting out call timeouts again, not cancelling"; + + host.provider().letGo(); + pump(200); +}