logos-protocol/cpp/implementations/plain/plain_logos_object.cpp

#include "plain_logos_object.h"

#include "logos_async_dispatch.h"
#include "qvariant_rpc_value.h"

#include <QCoreApplication>
#include <QDebug>
#include <QMetaObject>
#include <QTimer>
#include <QVariantMap>

#include <boost/asio/error.hpp>
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/strand.hpp>
#include <boost/system/error_code.hpp>

#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <functional>
#include <future>
#include <string>
#include <thread>
#include <utility>
#include <vector>

namespace logos::plain {

namespace {

// The default the deferred half of a call falls back to when the caller gave a
// non-positive timeout — the value awaitCompletion has always used, kept so the
// async path and the sync path give up at the same moment.
constexpr int kDeferredFallbackMs = 30000;

// 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");
}

// -----------------------------------------------------------------------------
// DeadlineService — the clock the per-call deadlines hang off. ONE thread for
// the whole process, and deliberately NOT the one the connections run on.
//
// WHY IT IS SEPARATE, which is the single most important decision in this file.
// Folding the per-call waiter thread away means the deadline has to live
// somewhere else, and the obvious somewhere — the connection's own strand, on
// IoContextPool::shared() — makes every deadline in the process hostage to that
// one io thread. It is not a theoretical hostage: this transport delivers user
// onEvent callbacks INLINE on the io thread (rpc_connection.h dispatchIncoming),
// and an event handler that calls another module is ordinary module code. A
// handler making a 2000ms synchronous call while a 200ms deadline is outstanding
// on a COMPLETELY DIFFERENT connection made that deadline fire at 2003ms;
// measured, and the reason this class exists. With the ambient timeouts in this
// stack — 5s in getMethods, 30s in the deferred fallback — a 300ms deadline
// becomes multi-second, and a handler that blocks forever means the deadline
// never fires at all. The whole point of a timeout is that it is the thing that
// still works when everything else is stuck.
//
// The three alternatives, and why not:
//
//   * A SECOND io thread in IoContextPool. Does not fix it — user handlers are
//     unbounded, so N simultaneously-blocked handlers need N+1 threads, and the
//     count is not knowable. It would also quietly break every "serialized by
//     there being one thread" assumption in the transport, which is a far larger
//     blast radius than this class.
//   * MOVING INLINE EVENT DELIVERY OFF THE STRAND (post user callbacks to the
//     Qt loop). Correct direction, much bigger change: it alters event ordering
//     and re-entrancy for every existing consumer of this transport, and it does
//     not help a deadline while the Qt loop itself is blocked.
//   * A Qt TIMER on the Qt event loop. Strictly worse than either: a
//     synchronous callMethod issued from the Qt thread — the most ordinary thing
//     a module does — blocks that loop for the whole call, so the deadline would
//     be hostage to exactly the calls it is supposed to bound.
//
// So: one dedicated thread, process-wide, that does nothing but arm, cancel and
// fire timers. It restores the independence the per-call waiter threads had, at
// one thread instead of one per pending call, which is the entire point of the
// fold. Nothing else may ever be posted here; user code reaches the Qt loop via
// postDelivery, and AsyncCall::deliver() is a flag, two map erases and a
// post.
// -----------------------------------------------------------------------------
class DeadlineService {
public:
    static DeadlineService& shared()
    {
        // Lazy, like IoContextPool::shared(): a process that never makes an
        // async plain call never starts this thread.
        static DeadlineService svc;
        return svc;
    }

    boost::asio::io_context& context() { return m_ioc; }

    DeadlineService(const DeadlineService&) = delete;
    DeadlineService& operator=(const DeadlineService&) = delete;

private:
    DeadlineService()
        : m_guard(boost::asio::make_work_guard(m_ioc))
        , m_thread([this] { m_ioc.run(); })
    {}

    ~DeadlineService()
    {
        m_guard.reset();
        m_ioc.stop();
        if (m_thread.joinable())
            m_thread.join();
    }

    boost::asio::io_context m_ioc;
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_guard;
    std::thread m_thread;
};

// The one place the choice above is made. Every per-call deadline in the process
// is armed on this context and nothing else is ever posted to it.
//
// The rejected design is one token different — IoContextPool::shared()
// .ioContext(), the connections' own thread — which is what makes the two tests
// in test_iofold.cpp that measure deadline accuracy under io-thread load worth
// having, and how they were validated. See that file for the numbers.
boost::asio::io_context& deadlineContext()
{
    return DeadlineService::shared().context();
}

// -----------------------------------------------------------------------------
// DeliveryService — the thread user callbacks land on in a process that has NO
// Qt event loop. A THIRD singleton thread, and the reasons it is neither of the
// other two are the whole design.
//
// WHAT WAS WRONG. The delivery hop below used to be exactly this:
//
//     QCoreApplication* app = QCoreApplication::instance();
//     if (!app) return;                       // <- the callback, dropped
//
// In a Qt host that branch only fires after QCoreApplication is gone, i.e.
// during shutdown, which is why it read as a reasonable guard. In a host that
// never had one it fires for EVERY call, forever: callMethodAsyncWithError,
// lp_invoke_async and every generated async wrapper promise their callback
// exactly once, and the plain transport — the transport whose entire reason for
// existing is to work without Qt — delivered zero. Not an error, not a
// timeout: silence, which turns a bounded call into an unbounded wait in every
// caller that awaits it. The header's guarantee was unconditional and untrue.
//
// WHAT IT IS NOT, which is the constraint that rules out the obvious fix.
// "Just call it inline when there is no loop" would run user code on whatever
// stack completed the call — the Asio read handler on the connection's strand,
// the deadline thread, or the middle of release(). That is the re-entrancy
// class this codebase has already paid for once: a deferred-multi completion
// running inline on the QtRO read stack, SIGSEGV, fixed by pushing it off the
// stack with QTimer::singleShot(0) (remote_transport.cpp). The whole point of
// this hop is that no user callback ever runs on an Asio handler stack, and a
// fix that delivers by removing the hop is not a fix.
//
// NOR IS IT THE DEADLINE THREAD. Posting user callbacks onto DeadlineService
// would make every deadline in the process hostage to user code — a callback
// that makes a synchronous call, or blocks, stops the clock for every other
// pending call. That is precisely the coupling DeadlineService was extracted to
// prevent, and its comment above says nothing else may ever be posted there.
// One thread each, therefore, and they cost nothing until used: a Qt host never
// starts this one at all.
//
// NOR IS IT "fail the call at issue time when there is no loop". That answer
// would make the plain transport unusable in exactly the deployment it was
// built for, and it still needs somewhere to deliver the failure it invents.
//
// ORDERING. A single thread draining a FIFO, so callbacks are delivered in the
// order they were resolved — the same property the Qt queued connection gives.
//
// AND IT IS NEVER DESTROYED, which is the one design decision in this class that
// is not obvious and is not free. The first cut made it an ordinary
// function-local static, and that is a use-after-free: a function-local static
// is constructed on FIRST DELIVERY, so anything with static storage that was
// constructed earlier — which is everything constructed before the first async
// call in the process — is destroyed LATER. Its destructor's delivery then posts
// into an io_context that has already run its own destructor, on a thread that
// has already been joined. Reproduced with nothing but the null-connection
// early-return path: three runs out of three, SIGSEGV under Guard Malloc inside
// scheduler::post_immediate_completion, reached from a static destructor through
// __cxa_finalize; and without Guard Malloc, silently, as delivered=0 — the very
// drop this class exists to fix, moved to a later moment.
//
// SO THE OBJECT OUTLIVES EVERY POSSIBLE CALLER, by never being destroyed at all
// and never registering a destructor to run. A deliberate, bounded leak: ONE
// io_context and ONE thread, in a process that is ending. What it buys is that
// postDelivery has no shutdown window — there is no state in which the vehicle
// is gone but callers remain — so a delivery issued from a static destructor
// after main() has returned still lands on a live delivery thread. Pinned by a
// probe that runs after main() in tests/protocol/test_delivery_without_qt.cpp.
//
// AND PROCESS EXIT MUST NOT WAIT FOR IT, which is why the thread is DETACHED
// rather than merely un-joined. exit() does not join threads, so an abandoned
// thread cannot hold the process open; detaching says so at the one place a
// future reader would otherwise reintroduce a join. Note what the old
// destructor's own comment was worried about — a user callback that blocks
// forever at static-destruction time hanging the join — and that this shape
// cannot have that failure at all, because there is no join. The hazard was
// always the reverse of what that comment described: the service died before its
// callers did.
//
// WHAT THE LEAK COSTS, stated rather than waved past. One thread lives from the
// first Qt-free async delivery to process exit, where it is sitting in the
// kernel waiting for work; and a callback that is RUNNING when the process exits
// can be cut off mid-flight, exactly as a Qt slot can be when the event loop's
// thread is torn down. What it cannot do is post into freed memory, which is
// what the alternative did.
class DeliveryService {
public:
    static DeliveryService& shared()
    {
        // Lazy, like IoContextPool::shared() and DeadlineService::shared(): a
        // process with a Qt event loop never constructs this and never starts
        // its thread. `new`, and never deleted: a function-local static of
        // OBJECT type would register a destructor with __cxa_atexit, which is
        // precisely the ordering hazard above. A pointer initialised once has no
        // destructor to register, so nothing about the moment of first use — the
        // middle of static destruction included — can leave a caller holding a
        // dead service.
        static DeliveryService* const svc = new DeliveryService();
        return *svc;
    }

    void post(std::function<void()> fn)
    {
        boost::asio::post(m_ioc, std::move(fn));
    }

    DeliveryService(const DeliveryService&) = delete;
    DeliveryService& operator=(const DeliveryService&) = delete;

private:
    DeliveryService()
        : m_guard(boost::asio::make_work_guard(m_ioc))
        , m_thread([this] { m_ioc.run(); })
    {
        // Detached at birth. Nothing will ever join it — see the note above —
        // and saying so here is what stops a future edit from adding a join to a
        // thread whose io_context is never stopped, which would hang exit()
        // forever rather than not at all.
        m_thread.detach();
    }

    // DELIBERATELY NOT DESTRUCTIBLE. The compiler now enforces what the comment
    // asks for: no `static DeliveryService svc;`, no unique_ptr, no `delete`,
    // and therefore no way to reintroduce a destroyed-before-its-callers
    // service by accident. `new` needs no destructor, so shared() still
    // compiles.
    ~DeliveryService() = delete;

    boost::asio::io_context m_ioc;
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_guard;
    std::thread m_thread;
};

// Hand `callback(result, err)` over to a LATER stack, so PlainLogosObject's
// async path matches LogosObject's interface contract: callbacks are never
// delivered synchronously inside the call that issued them, and never on the
// transport's io thread.
//
// TWO VEHICLES, one rule. When this process runs a Qt event loop the callback
// is queued onto it exactly as before, so it lands on the Qt thread and cannot
// race QObjects or UI code — unchanged behaviour for every Qt host. When this
// process has never had one, the callback goes to the dedicated delivery thread
// above instead of being dropped. Nothing in between: there is no path that
// runs it on the completing thread.
//
// WHY "HAS NEVER HAD ONE" AND NOT "DOES NOT HAVE ONE RIGHT NOW", which is the
// one subtlety in this function. instance() also goes null during
// ~QCoreApplication, so the naive test would start delivering on a foreign
// thread while the Qt app is being torn down — and module teardown after the
// application is gone is not exotic, it is what static-destruction ordering
// produces, with stopAndCancelCalls() handing every in-flight call a
// cancellation callback at exactly that moment. Running user code on a side
// thread into half-destroyed module state is a worse outcome than not running
// it, and it would be a NEW failure mode introduced into every Qt app by a
// bug-fix change. So the flag latches: a process that has ever been seen with
// an event loop keeps the old shutdown behaviour (the callback is dropped),
// and a process that has never had one gets the delivery thread. A Qt host
// therefore sees no behavioural difference at all — which is the point.
//
// THE RESIDUE, ENUMERATED, because "exactly once" was untrue here once already
// and a vague replacement is how that happens twice. Three shapes do NOT get a
// delivery, and all three are properties of the host process, not of the call:
//
//   1. A Qt process AFTER ~QCoreApplication. The callback is dropped, by the
//      decision above.
//   2. A process that constructs a QCoreApplication and never runs its event
//      loop. The delivery is queued onto a loop that never turns, so it never
//      runs. Nothing here can fix that: the callback's whole contract is that it
//      arrives on the Qt thread, and delivering it anywhere else would be the
//      thread-affinity violation this hop exists to prevent. Not new, not
//      introduced here — this is what "queue it onto the Qt loop" has always
//      meant — but it is a case where the header used to promise exactly once,
//      so it is named.
//   3. A process that had a QCoreApplication TRANSIENTLY — constructed and
//      destroyed while the process goes on to run Qt-free. The latch keeps
//      dropping for the rest of that process's life. It is the price of the
//      choice in (1): from inside postDelivery, "the app is gone because the
//      process is shutting down" and "the app is gone because a helper's app
//      object went out of scope" are the same observation, and getting (1) wrong
//      breaks every existing Qt host at exit while getting (3) wrong breaks a
//      shape no host in this codebase has. The trade is deliberate and is not
//      hidden: the guarantee below is worded as "a process with no
//      QCoreApplication in its life", not "a process without one right now".
//
// logos_object.h states all three where callers read the contract.
//
// Deliberately a FREE function taking everything BY VALUE, and deliberately not
// a member: the queued lambda runs later, which for a call cancelled by
// teardown is after the PlainLogosObject is already gone. Nothing it touches
// may belong to the object — which is why AsyncCall copies objectName/method up
// front instead of reading m_objectName from inside here. Do not give this a
// `this`.
//
// THE FOLD MADE THIS LOAD-BEARING TWICE OVER. It was already the reason a
// delivery could outlive the handle. It is now also the answer to the
// re-entrancy hazard: every one of the four places that can complete a call —
// the Asio read handler on the connection's strand, fail()'s sweep on an
// arbitrary thread, the deadline handler on the timer thread, and teardown on
// the caller's thread — routes its delivery through here, so NO user callback
// ever runs on an Asio handler stack. That is the class of bug that produced the
// deferred-multi SIGSEGV on the QtRO twin, whose fix (remote_transport.cpp) is
// the same move by a different vehicle: QTimer::singleShot(0).
// Latched the first time an event loop is seen; never cleared. See postDelivery.
std::atomic<bool> g_processHadQtLoop{false};

void postDelivery(PlainLogosObject::AsyncResultErrorCallback callback,
                  QVariant result, logos::CallError err)
{
    if (QCoreApplication* app = QCoreApplication::instance()) {
        g_processHadQtLoop.store(true, std::memory_order_relaxed);
        QMetaObject::invokeMethod(app,
            [callback = std::move(callback), result = std::move(result),
             err = std::move(err)]() mutable {
                callback(result, err);
            },
            Qt::QueuedConnection);
        return;
    }
    // The application existed and is gone: this is shutdown, and the old
    // behaviour — drop it — is deliberately kept. See above.
    if (g_processHadQtLoop.load(std::memory_order_relaxed)) return;

    DeliveryService::shared().post(
        [callback = std::move(callback), result = std::move(result),
         err = std::move(err)]() mutable {
            callback(result, err);
        });
}

} // anonymous namespace

// -----------------------------------------------------------------------------
// AsyncCall — one in-flight asynchronous call. THIS IS WHAT REPLACED THE THREAD.
//
// The old design gave every pending RPC an OS thread whose only job was to be
// blockable: std::future cannot be waited on with a deadline AND a cancel, so
// the waiter polled it in 25ms slices, then parked on a condition variable for
// the deferred half, then delivered. Three costs came with that — a thread per
// pending call, a 25ms floor on teardown, and a registry-plus-reaping protocol
// to stop finished threads accumulating (a thread cannot join itself).
//
// Here the call is a piece of STATE that three events race to finish:
//
//   * the reply, delivered by RpcConnection as a handler (its strand, or any
//     thread via fail(), or inline when the connection is already stopped);
//   * the deadline, an asio::steady_timer on the DeadlineService's own thread;
//   * cancellation, from teardown on an arbitrary thread.
//
// EXACTLY ONCE is the `claim()` CAS below, and nothing else. That is a real
// change of mechanism and the thing most worth distrusting: the old guarantee
// was structural (one thread, one function, four returns, and a join proving it
// had finished), whereas three independent callers can all arrive here. The CAS
// is what makes the first one win and the other two no-ops, on every path.
//
// LIFETIME is ownership, not a barrier. Each of those three holds a shared_ptr
// to this object; the state it needs on the HANDLE is reached through a weak_ptr
// to CallState, and the connection through a weak_ptr too. Nothing here
// dereferences the PlainLogosObject, so release()'s `delete this` is none of its
// business and teardown has nothing to wait for.
// -----------------------------------------------------------------------------
struct AsyncCall : std::enable_shared_from_this<AsyncCall> {
    using clock = std::chrono::steady_clock;

    AsyncCall(std::weak_ptr<PlainLogosObject::CallState> st,
              std::weak_ptr<RpcConnectionBase> cn,
              std::uint64_t callNumber,
              std::string obj, std::string meth, int timeout,
              PlainLogosObject::AsyncResultErrorCallback cb)
        // A strand of its own over the shared deadline thread. With one thread
        // in that service the strand is redundant today; it is here so that
        // "every touch of this timer is serialized" stays a property of the
        // code rather than of the thread count, because asio timers are
        // "Shared objects: Unsafe" and a second service thread would otherwise
        // turn a re-arm racing its own handler into undefined behaviour.
        : timer(boost::asio::make_strand(deadlineContext()))
        , state(std::move(st))
        , conn(std::move(cn))
        , id(callNumber)
        , objectName(std::move(obj))
        , method(std::move(meth))
        , timeoutMs(timeout)
        , callback(std::move(cb))
    {}

    boost::asio::steady_timer                  timer;
    std::weak_ptr<PlainLogosObject::CallState> state;
    // Only ever used to withdraw this call's registration from the connection's
    // pending map — see cancelPending(). weak, because the connection outlives
    // the handle but not necessarily this call's last handler.
    std::weak_ptr<RpcConnectionBase>           conn;
    const std::uint64_t                        id;
    const std::string                          objectName;
    const std::string                          method;
    const int                                  timeoutMs;

    // Set under CallState::mu when a "multi" provider defers, read under it in
    // deliver() — the one field two threads can reach.
    QString                                    callId;

    std::mutex                                          cbMu;
    PlainLogosObject::AsyncResultErrorCallback          callback;
    std::atomic<bool>                                   delivered{false};

    // ── the exactly-once gate ────────────────────────────────────────────────
    //
    // Three independent callers race for the right to resolve a call — the
    // reply handler, the deadline and teardown — and exactly one may reach the
    // user's callback. Two halves, INDEPENDENTLY SUFFICIENT, which is worth
    // recording because it means neither is redundant: the CAS is the one that
    // also skips the registry erase, the pending withdrawal and the timer
    // cancel, and the swap is what makes the callback itself unrepeatable.
    //
    // This replaced a structural guarantee — one waiter thread, one function
    // body, and a join proving it had finished — so it is the guarantee in this
    // file most worth distrusting, and the two tests that actually detect its
    // absence are named in tests/protocol/CMakeLists.txt. The per-path
    // exactly-once assertions are NOT among them: a call resolved once calls
    // deliver() once whatever guards it.
    bool claim()
    {
        bool expected = false;
        return delivered.compare_exchange_strong(expected, true,
                                                 std::memory_order_acq_rel);
    }

    PlainLogosObject::AsyncResultErrorCallback takeCallback()
    {
        std::lock_guard<std::mutex> g(cbMu);
        PlainLogosObject::AsyncResultErrorCallback cb;
        cb.swap(callback);
        return cb;
    }

    // Idempotent by construction: every later caller returns without touching
    // the callback, the connection or the timer.
    void deliver(QVariant value, logos::CallError err)
    {
        // LEAVE THE HANDLE'S REGISTRIES FIRST — before the exactly-once gate,
        // and unconditionally, which is not where it reads most naturally.
        //
        // A duplicate deliver() has something to clean up. A provider can
        // answer the pending sentinel AFTER this call's deadline has already
        // passed: the timer resolves the call, and only then does the reply
        // handler arrive, see a sentinel, and file this AsyncCall under
        // CallState::deferred. Behind the gate, that entry would never be taken
        // out again — one leaked map entry per slow-sentinel call, for the life
        // of the handle, which is precisely the retention the fold exists to
        // fix. (The reply handler also checks `delivered` before filing, so this
        // only has to cover the instant between that check and the write; the
        // re-armed deadline is what eventually runs this erase.)
        //
        // Erasing twice is free: ids come from RpcConnection::nextId() and are
        // unique per call, so this can never take out somebody else's entry.
        if (auto st = state.lock()) {
            std::lock_guard<std::mutex> g(st->mu);
            st->inflight.erase(id);
            if (!callId.isEmpty()) st->deferred.erase(callId);
        }

        if (!claim()) return;

        PlainLogosObject::AsyncResultErrorCallback cb = takeCallback();

        // AND LEAVE THE CONNECTION'S. m_pendingCalls is erased by exactly two
        // events on its own — a decoded reply with this id, and fail()'s
        // teardown sweep — so a call resolved by its DEADLINE, or by teardown of
        // the handle rather than of the connection, used to leave its
        // registration there for the whole life of the connection, which
        // outlives every handle it hands out. That was true of the promise this
        // replaced too; it is closed here rather than inherited. When the reply
        // IS what got us here, dispatchIncoming has already erased it and this
        // is a lookup that finds nothing.
        if (auto c = conn.lock()) c->cancelPending(id);

        cancelTimer();
        // Last, and never with a lock held: the delivery hop.
        if (cb) postDelivery(std::move(cb), std::move(value), std::move(err));
    }

    // Callable from ANY thread: the arm is POSTED onto the timer's own strand,
    // so the timer object itself is only ever touched from the deadline thread.
    //
    // `when` is an ABSOLUTE deadline computed by the caller, not a duration, so
    // the post hop cannot stretch it — the timer fires when the caller said it
    // would even if the deadline thread is momentarily busy. `reportMs` is only
    // what the timeout REPORTS; it diverges from the wall time for the deferred
    // half of a non-positive-timeout call, which falls back to 30s the way
    // awaitCompletion always has.
    void armTimer(clock::time_point when, int reportMs)
    {
        auto self = shared_from_this();
        boost::asio::post(timer.get_executor(), [self, when, reportMs] {
            self->timer.expires_at(when);
            self->timer.async_wait([self, reportMs](const boost::system::error_code& ec) {
                if (ec == boost::asio::error::operation_aborted) return;
                self->deliver(QVariant(),
                              logos::callErrorTimeout(self->objectName,
                                                      self->method, reportMs));
            });
        });
    }

    // Also posted, for the same reason. Cancelling matters for retention rather
    // than correctness (the CAS already makes a late timeout a no-op): without
    // it, a call answered in 1ms with a 20s timeout would keep this object alive
    // for the remaining 19.999s.
    void cancelTimer()
    {
        auto self = shared_from_this();
        boost::asio::post(timer.get_executor(), [self] {
            try { self->timer.cancel(); } catch (...) {}
        });
    }
};

// -----------------------------------------------------------------------------
// The live-reference count that makes release() safe against a call already
// running, and the detector beside it for the shapes it cannot save. Two
// counters, a flag and a printf; the reasoning is the part worth reading, and it
// lives over release().
// -----------------------------------------------------------------------------

// This thread's nesting depth inside ANY PlainLogosObject entry point. Two
// readers, both of them detector-side: it subtracts the caller's own frames from
// the count reportConcurrentCallers reads, and it tells EntryGuard whether an
// entry is a FRESH one from outside or a nested internal one, which is what keeps
// the "entered after release" report off callMethodWithError ->
// ensureCompletionSub -> onEvent.
//
// THE SHAPE IT EXISTS TO IGNORE: a thread that is inside a public method and,
// from there, reaches teardown on the same object — a callback invoked from
// within a call, releasing the handle it was called through. That is a
// well-defined single-threaded sequence, not the race, and it is SAFE under the
// reference count (the outer call holds a reference, so release() cannot destroy
// the object underneath it and the guard does it on the way out). Reporting it
// would be a false alarm. No path in this file invokes user code from inside a
// guarded method today, so this is guarding a shape that does not yet exist —
// which is exactly when it is cheap to guard.
//
// It counts entries into ANY object rather than into a particular one, because
// a per-object thread-local is a map lookup on every call and the difference
// only matters for a thread that is inside object A while releasing object B.
// That case is not the defect, and the bias it introduces is the safe one:
// this can only ever make the detector MISS a real race, never invent one.
// Internal linkage: nothing outside this file may read it, and it must not
// become an exported thread-local in the shared build.
static thread_local int t_entryDepth = 0;

// THE REFERENCE IS THE FIRST AND THE LAST THING EITHER OF THESE TOUCHES, and
// that ordering is the whole safety argument.
//
// ON THE WAY IN the reference must be taken before any other access to the
// object, because everything after it is what the reference protects. On the way
// out it must be dropped after every other access, because dropping it may
// DESTROY the object — this guard is the thing that performs the final `delete`
// when it is the last one out, and any bookkeeping after that point would be a
// store into freed memory. The earlier, detector-only version of this guard
// learned that the hard way: it restored m_lastEntryPoint after decrementing,
// which opened a window exactly one store wide for a concurrent release() to
// `delete this` in, and that segfaulted a Linux CI run in an existing test whose
// io-thread event callback releases the handle (IoFoldTest
// .ReleaseFromInsideAnIoThreadEventCallbackDoesNotWedge).
//
// So: reference first, reference last, everything else in between.
//
// THE RESIDUAL INSTANT is the increment itself, and it is the boundary of what
// this mechanism can do. A caller whose increment lands after release() has
// already dropped the owner's reference and freed the storage is incrementing an
// integer inside a dead object; nothing can be read out of freed memory to
// prevent that, which is why entering after release() stays a caller error
// rather than becoming safe. A call that entered BEFORE release() was called has
// no such problem: its increment is ordered before release()'s decrement in the
// modification order of m_liveRefs, so release() sees it, declines to destroy
// and leaves the destruction to this guard.
//
// The cost of handing the diagnostic name back before the decrement is a
// slightly vaguer message: a concurrent reader can see the OUTER frame's name
// rather than the inner one. That is a diagnostic string; correctness of the
// count is what matters.
PlainLogosObject::EntryGuard::EntryGuard(PlainLogosObject* obj,
                                         const char* entryPoint)
    : m_obj(obj)
    , m_prev(nullptr)
{
    ++t_entryDepth;                                        // thread-local
    m_obj->m_liveRefs.fetch_add(1, std::memory_order_acq_rel);        // FIRST
    m_obj->m_callsInFlight.fetch_add(1, std::memory_order_acq_rel);
    m_prev = m_obj->m_lastEntryPoint.exchange(entryPoint,
                                              std::memory_order_relaxed);
    // A FRESH entry — not a nested one — into an object whose owner has already
    // released it. Safe only because somebody else's reference is still holding
    // the storage alive, which is luck and not a contract, so it is named.
    //
    // The depth test excludes the internal nested paths (callMethodWithError ->
    // ensureCompletionSub -> onEvent), whose outer frame entered long before
    // release() and which are not a second user of the handle. It is DEFENSIVE
    // rather than load-bearing today: removing it leaves the whole suite green,
    // because no nested entry currently happens after teardown has started —
    // ensureCompletionSub runs before the call parks, not after it wakes. It is
    // kept because that is a property of today's call paths and not of this
    // check, and a false alarm here aborts a correct program.
    if (t_entryDepth == 1 && m_obj->m_released.load(std::memory_order_acquire))
        m_obj->reportEntryAfterRelease(entryPoint);
}

PlainLogosObject::EntryGuard::~EntryGuard()
{
    // Hand the name back first, so a nested entry does not leave the diagnostic
    // blaming onEvent() for a caller that is really parked in
    // callMethodWithError(). Across threads this is last-writer-wins and
    // therefore approximate — it names ONE of the callers in flight, which is
    // all the message claims.
    m_obj->m_lastEntryPoint.store(m_prev, std::memory_order_relaxed);
    --t_entryDepth;                                        // thread-local
    m_obj->m_callsInFlight.fetch_sub(1, std::memory_order_acq_rel);
    // LAST, and it may be a `delete`. Read into a local first: after the
    // fetch_sub this guard may no longer have an object to name, and after the
    // delete it must touch nothing at all — which is why this is the final
    // statement of the final destructor of every entry point. EntryGuard is
    // declared FIRST in each of them, so it is destroyed last, after every other
    // local and after the return value has been constructed in the caller's
    // storage.
    PlainLogosObject* obj = m_obj;
    if (obj->m_liveRefs.fetch_sub(1, std::memory_order_acq_rel) == 1)
        delete obj;                    // the last one out turns the lights off
}

void PlainLogosObject::reportConcurrentCallers(const char* where) const
{
    // `- t_entryDepth` removes this thread's own frames; see t_entryDepth.
    const int others =
        m_callsInFlight.load(std::memory_order_acquire) - t_entryDepth;
    if (others <= 0) return;

    const char* entry = m_lastEntryPoint.load(std::memory_order_relaxed);

    // Written straight to stderr, not through qWarning/qCritical, on purpose.
    // By the time this fires the program has already committed the error and
    // may be one instruction from a SIGSEGV, so the message must not depend on
    // a Qt message handler being installed, being reachable, or flushing — and
    // must not depend on a QCoreApplication existing at all, which on this
    // transport it may not.
    std::fprintf(stderr,
        "\nLOGOS FATAL: PlainLogosObject::%s on '%s' ran while %d call(s) from "
        "other threads are still inside this object (most recent entry: "
        "PlainLogosObject::%s).\n"
        "  This path destroys the object NOW, so those calls are about to use "
        "freed memory. release() would have been safe — it drops the owner's "
        "reference and lets the LAST call in flight do the destroying — but a "
        "direct `delete` has nobody to hand the destruction to. Use release(), "
        "or wait for the calls.\n"
        "  This is a diagnostic, not a rescue: the object is destroyed either "
        "way. See the note over PlainLogosObject::release().\n",
        where, m_objectName.c_str(), others,
        entry ? entry : "<unknown>");
    std::fflush(stderr);

    // Debug builds stop here, so the misuse is a named abort at the line that
    // committed it rather than a SIGSEGV somewhere else. Release builds carry
    // on into the same crash they would have had, with a log line naming the
    // cause — the point being that turning a shipped app's latent misuse into a
    // hard abort is not a decision a bug-fix release gets to make for its
    // consumers.
#ifndef NDEBUG
    std::abort();
#endif
}

// The other half of the diagnostic: not "teardown found a caller inside", but "a
// caller came in after teardown". Reachable only while somebody else's reference
// is still holding the storage alive — if it were not, this function would be
// running on freed memory and there would be nothing to report with. That is
// exactly why the message says the program is out of contract even though it did
// not crash this time.
void PlainLogosObject::reportEntryAfterRelease(const char* entryPoint) const
{
    std::fprintf(stderr,
        "\nLOGOS FATAL: PlainLogosObject::%s was entered on '%s' AFTER "
        "release() was called on the same handle.\n"
        "  release() means the handle is finished: it is only still readable "
        "because another call in flight is holding the object alive, and with "
        "that call gone this would have been a use-after-free instead of a "
        "message. A LogosObject handle is safe to use from several threads at "
        "once, and calls already inside it are safe against release() — but "
        "STARTING a call after release() is never safe.\n"
        "  See the note over PlainLogosObject::release().\n",
        entryPoint, m_objectName.c_str());
    std::fflush(stderr);

#ifndef NDEBUG
    std::abort();
#endif
}

PlainLogosObject::PlainLogosObject(std::string objectName,
                                   std::shared_ptr<RpcConnectionBase> conn)
    : m_objectName(std::move(objectName))
    , m_conn(std::move(conn))
{
}

PlainLogosObject::~PlainLogosObject()
{
    // BEFORE anything is torn down, and it still has a job to do even now that
    // release() is safe. Two ways to get here:
    //
    //   * through the reference count — release() dropped the owner's reference,
    //     or the last call in flight dropped its own. Then no call is inside the
    //     object by construction, the count is zero, and this returns silently.
    //   * through `delete obj` on a handle somebody kept a second pointer to,
    //     which bypasses the reference count entirely and destroys the object
    //     while those calls are still running. Nothing here can defer THAT — the
    //     caller has already demanded the storage back — so it is reported, and
    //     aborts in a debug build, which is the whole reason this call survived
    //     the change that made release() safe.
    reportConcurrentCallers("~PlainLogosObject()");
    // Impl, not the guarded entry point: taking an EntryGuard here would raise
    // the reference count from zero and drop it again, and a drop to zero
    // destroys the object — recursing into `delete this` from inside the
    // destructor. Same reason in release().
    disconnectEventsImpl();
    stopAndCancelCalls();
}

void PlainLogosObject::stopAndCancelCalls()
{
    std::vector<std::shared_ptr<AsyncCall>> outstanding;
    {
        // The flag is published under the same mutex awaitCompletion evaluates
        // its predicate under, so the parked synchronous caller cannot read
        // `false`, decide to sleep, and only then miss the notify_all below.
        std::lock_guard<std::mutex> g(m_state->mu);
        m_state->stopping.store(true, std::memory_order_release);
        outstanding.reserve(m_state->inflight.size());
        for (auto& entry : m_state->inflight)
            outstanding.push_back(entry.second);
        m_state->inflight.clear();
        m_state->deferred.clear();
        // Buffered completions nobody can claim any more. They are dropped here
        // rather than left to the state block's own destruction so that a
        // handler still holding a share of it does not keep them alive.
        m_state->completions.clear();
    }
    m_state->cv.notify_all();

    // Cancelled OUTSIDE the lock, because deliver() takes it to erase its own
    // registry entries. (It will find nothing — they were just cleared — which
    // is fine and is why this cannot deadlock either way.)
    //
    // A cancelled call still DELIVERS, exactly once. Returning silently 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.
    for (auto& call : outstanding)
        call->deliver(QVariant(), callErrorReleased(m_objectName, call->method));

    // And that is the whole of teardown. NOTHING IS WAITED FOR: no thread to
    // join, no io-thread barrier. The reply handler and the deadline handler for
    // a cancelled call may still be queued; each holds its own shared_ptr to the
    // AsyncCall, finds the gate already taken, and drops its share. None of them
    // can reach this object, so it may be deleted the instant this returns.
    //
    // THE BARRIER THAT LOOKS RIGHT AND ISN'T: "post a no-op onto the connection
    // strand and wait for it" would prove no handler is mid-flight, and it
    // wedges the process — IoContextPool runs exactly ONE thread, this transport
    // delivers user event callbacks inline on it, and release()-from-an-event-
    // callback is shipped behaviour (remote_transport.cpp), so the caller can BE
    // the only thread that could drain the barrier. Tried, deadlocked,
    // discarded; ownership is what replaced the join, not a barrier.
    // test_iofold.cpp pins the reentrant case with a watchdog.
}

QVariant PlainLogosObject::callMethod(const QString& authToken,
                                      const QString& methodName,
                                      const QVariantList& args,
                                      int timeoutMs)
{
    // Adapter over the error-carrying implementation: discards the diagnosis,
    // which is exactly what this entry point has always done.
    return callMethodWithError(authToken, methodName, args, timeoutMs, nullptr);
}

QVariant PlainLogosObject::callMethodWithError(const QString& authToken,
                                               const QString& methodName,
                                               const QVariantList& args,
                                               int timeoutMs,
                                               logos::CallError* err)
{
    EntryGuard guard(this, "callMethodWithError()");
    if (err) err->clear();
    if (!m_conn || !m_conn->isOpen()) {
        if (err)
            *err = logos::callErrorTransport(
                m_objectName, "connection to '" + m_objectName + "' is not open");
        return QVariant();
    }

    // Subscribe to the completion channel BEFORE sending, so a "multi" provider's
    // completion can't race ahead of the waiter (it's buffered either way).
    ensureCompletionSub();

    CallMessage msg;
    msg.id        = m_conn->nextId();
    msg.authToken = authToken.toStdString();
    msg.object    = m_objectName;
    msg.method    = methodName.toStdString();
    msg.args      = qvariantListToRpcList(args);

    const std::uint64_t callNumber = msg.id;
    auto fut = m_conn->sendCall(std::move(msg));

    if (fut.wait_for(std::chrono::milliseconds(timeoutMs)) != std::future_status::ready) {
        qWarning() << "PlainLogosObject::callMethod: timeout for" << methodName;
        // Withdraw the registration this call left in the connection. The sync
        // path has the same orphan the async one does — nothing erases a
        // pending entry whose reply never comes — and the promise behind it
        // holds a future nobody will ever read again.
        m_conn->cancelPending(callNumber);
        if (err)
            *err = logos::callErrorTimeout(m_objectName, methodName.toStdString(),
                                           timeoutMs);
        return QVariant();
    }
    auto res = fut.get();
    if (!res.ok) {
        qWarning() << "PlainLogosObject::callMethod:" << methodName
                   << "failed:" << QString::fromStdString(res.err);
        // res.errCode / res.err have been on the wire since the plain transport
        // existed; this is the first caller to keep them. MODULE_NOT_LOADED in
        // particular is how "the module isn't there" reaches us on this
        // transport — requestObject never checks publication — so without this
        // the single most common failure was reported as a null result.
        if (err)
            *err = logos::callErrorFromWire(m_objectName, res.errCode, res.err);
        return QVariant();
    }
    const QVariant value = rpcValueToQVariant(res.value);
    // A "multi" provider may have deferred: it returned a pending sentinel and
    // pushes the real result as a completion event. Wait for it, keyed by callId.
    {
        QString callId;
        if (logos::isPendingCallSentinel(value, &callId))
            return awaitCompletion(callId, timeoutMs, methodName, err);
    }
    return value;
}

void PlainLogosObject::ensureCompletionSub()
{
    // Fast path. Every call after the first pays one acquire load and nothing
    // else — and inherits, through it, the ordering the first caller
    // established (see the header).
    if (m_completionSubscribed.load(std::memory_order_acquire)) return;

    // Serializing is the whole fix. A second caller that arrives while the
    // first is still inside subscribeToCompletions() BLOCKS here instead of
    // racing ahead with a Call the provider can answer before the Subscribe
    // frame has been enqueued.
    std::call_once(m_completionSubOnce, [this] {
        subscribeToCompletions();
        // Published last, so the fast path above cannot let a caller through on
        // a subscription that is not yet on the strand.
        m_completionSubscribed.store(true, std::memory_order_release);
    });
}

void PlainLogosObject::subscribeToCompletions()
{
    // Reuse the normal event subscription path (tracked in m_subs, so
    // disconnectEvents() tears it down). The handler fires on the connection's
    // IO thread.
    //
    // It captures a weak_ptr to CallState and nothing else — in particular NOT
    // `this`. That unsubscribe is real (RpcConnection::sendUnsubscribe erases the
    // entry under the connection's mutex) but it is not enough on its own:
    // dispatchIncoming copies the handler out under that mutex and invokes it
    // with the mutex dropped, so an erase racing an already-copied handler
    // changes nothing about the invocation in flight, and nothing joins the io
    // thread. With `this` captured, a completion arriving across a release()
    // wrote into freed memory — see test_plain_completion_sub_lifetime.cpp.
    //
    // weak, not shared, deliberately: locking is what keeps the block alive for
    // the length of one callback, and failing to lock is what makes a handler
    // that outlives its owner — for this reason or any future one — a no-op
    // instead of an append to a map nobody will ever drain.
    std::weak_ptr<CallState> weak = m_state;
    onEvent(logos::callCompleteEvent(), [weak](const QString&, const QVariantList& data) {
        if (data.size() != 2) return;
        const std::shared_ptr<CallState> st = weak.lock();
        if (!st) return;              // the handle and its state are both gone
        const QString callId = data.at(0).toString();

        std::shared_ptr<AsyncCall> call;
        {
            std::lock_guard<std::mutex> g(st->mu);
            auto it = st->deferred.find(callId);
            if (it != st->deferred.end()) {
                call = it->second;
                st->deferred.erase(it);
            } else {
                // Nobody is waiting on it yet: either a SYNCHRONOUS caller is
                // about to park on it, or it arrived before its own sentinel
                // was recorded. Buffer it, exactly as before.
                st->completions[callId] = data.at(1);
            }
        }
        if (call) {
            // Resolves the async call HERE, on the io thread — but deliver()
            // only takes a flag, drops two map entries and posts, so the user's
            // callback still runs on the Qt loop. Called with st->mu released:
            // deliver() takes it.
            call->deliver(data.at(1), logos::CallError{});
            return;
        }
        st->cv.notify_all();
    });
}

QVariant PlainLogosObject::awaitCompletion(const QString& callId, int timeoutMs,
                                           const QString& methodName,
                                           logos::CallError* err)
{
    // A LOCAL SHARE of the state, held for the whole wait. This function only
    // ever runs on the SYNCHRONOUS caller's own thread, so that caller cannot
    // be releasing the handle underneath it — but taking the share costs one
    // atomic increment and removes the question entirely.
    const std::shared_ptr<CallState> st = m_state;
    std::unique_lock<std::mutex> lk(st->mu);
    const auto effectiveMs = timeoutMs > 0 ? timeoutMs : kDeferredFallbackMs;
    const auto deadline = std::chrono::steady_clock::now()
        + std::chrono::milliseconds(effectiveMs);
    // Interruptible by construction: widen the predicate, and
    // stopAndCancelCalls()' notify_all does the rest. No slicing, so no latency
    // floor at all here — a stop wakes this wait immediately.
    st->cv.wait_until(lk, deadline, [&] {
        return st->completions.count(callId) > 0
            || st->stopping.load(std::memory_order_relaxed);
    });

    // AN ANSWER ALREADY IN HAND BEATS A CONCURRENT STOP: there is a real result
    // here, so hand it over rather than manufacture a failure that did not
    // happen. Callers re-acquire, retry and log on transport_error.
    const auto it = st->completions.find(callId);
    if (it != st->completions.end()) {
        const QVariant result = it->second;
        st->completions.erase(it);
        return result;
    }
    if (st->stopping.load(std::memory_order_relaxed)) {
        qWarning() << "PlainLogosObject: deferred call" << callId
                   << "abandoned — object released while it was in flight";
        if (err)
            *err = callErrorReleased(m_objectName, methodName.toStdString());
        return QVariant();
    }
    qWarning() << "PlainLogosObject: deferred call" << callId << "timed out";
    if (err)
        *err = logos::callErrorTimeout(m_objectName, methodName.toStdString(),
                                       effectiveMs);
    return QVariant();
}

void PlainLogosObject::callMethodAsync(const QString& authToken,
                                       const QString& methodName,
                                       const QVariantList& args,
                                       int timeoutMs,
                                       AsyncResultCallback callback)
{
    // Adapter over the error-carrying implementation: discards the diagnosis,
    // which is exactly what this entry point has always done.
    if (!callback) return;
    callMethodAsyncWithError(authToken, methodName, args, timeoutMs,
        [cb = std::move(callback)](QVariant v, const logos::CallError&) mutable {
            cb(std::move(v));
        });
}

void PlainLogosObject::callMethodAsyncWithError(const QString& authToken,
                                                const QString& methodName,
                                                const QVariantList& args,
                                                int timeoutMs,
                                                AsyncResultErrorCallback callback)
{
    EntryGuard guard(this, "callMethodAsyncWithError()");
    if (!callback) return;
    if (!m_conn || !m_conn->isOpen()) {
        // Defer even the failure path — LogosObject's contract requires
        // callbacks on a subsequent event-loop iteration, never inline.
        postDelivery(std::move(callback), QVariant(),
                          logos::callErrorTransport(
                              m_objectName,
                              "connection to '" + m_objectName + "' is not open"));
        return;
    }

    ensureCompletionSub();

    CallMessage msg;
    msg.id        = m_conn->nextId();
    msg.authToken = authToken.toStdString();
    msg.object    = m_objectName;
    msg.method    = methodName.toStdString();
    msg.args      = qvariantListToRpcList(args);

    // Copied, not read from the object later: everything below this line may
    // outlive the handle.
    const std::uint64_t callNumber = msg.id;
    const std::string   objectName = m_objectName;
    const std::string   method     = methodName.toStdString();

    // ── the call, as state rather than as a thread ──────────────────────────
    //
    // There used to be a std::thread here whose entire job was to be blockable,
    // and a TODO saying to fold it into the io_context the connection already
    // runs on. This is that fold. Nothing below spawns, joins, sleeps or polls.
    auto call = std::make_shared<AsyncCall>(m_state, m_conn, callNumber,
                                            objectName, method, timeoutMs,
                                            std::move(callback));

    // The deadline is fixed HERE, before the send, and as an absolute point —
    // so neither the post onto the timer thread nor anything the io thread is
    // doing can stretch what the caller asked for.
    call->armTimer(AsyncCall::clock::now() + std::chrono::milliseconds(timeoutMs),
                   timeoutMs);

    bool refused = false;
    {
        std::lock_guard<std::mutex> g(m_state->mu);
        // Same branch, same honesty as before the fold: stopping is raised only
        // by teardown, so a caller that can read it as true here is already
        // calling a method on an object whose destructor is running — this very
        // load is the use-after-free, and nothing in this function can repair
        // that. It is kept because failing this way — one callback, with the
        // error a cancelled call gets — is strictly better than registering a
        // call nobody will ever cancel.
        //
        // What HAS changed is the blast radius if it ever became reachable: the
        // state it would leak an entry into is shared-owned and would simply
        // outlive the handle, instead of being a thread nobody joins.
        if (m_state->stopping.load(std::memory_order_acquire))
            refused = true;
        else
            m_state->inflight.emplace(callNumber, call);
    }
    // OUTSIDE the lock, and that is not a stylistic preference: deliver() takes
    // CallState::mu to leave the registries, and this mutex is not recursive.
    // Delivering from inside the scope above self-deadlocks — on the one branch
    // whose whole purpose is to fail gracefully.
    if (refused) {
        call->deliver(QVariant(), callErrorReleased(objectName, method));
        return;
    }

    // The reply. Handed over by RpcConnection as it arrives, on its strand for
    // the normal path — and on an arbitrary thread from fail(), or inline right
    // here if the connection is already stopped. All three are fine: every exit
    // below funnels into AsyncCall::deliver, whose CAS makes the first one win
    // and which hops to the Qt loop rather than running the user's callback on
    // whatever stack it happens to be on.
    std::weak_ptr<CallState> weakState = m_state;
    m_conn->sendCallAsync(std::move(msg), [call, weakState](ResultMessage res) {
        if (!res.ok) {
            call->deliver(QVariant(),
                          logos::callErrorFromWire(call->objectName, res.errCode,
                                                   res.err));
            return;
        }
        QVariant value = rpcValueToQVariant(res.value);
        QString callId;
        if (!logos::isPendingCallSentinel(value, &callId)) {
            call->deliver(std::move(value), logos::CallError{});
            return;
        }

        // ── the deferred ("multi") half ─────────────────────────────────────
        auto st = weakState.lock();
        if (!st) {
            call->deliver(QVariant(),
                          callErrorReleased(call->objectName, call->method));
            return;
        }

        // Already resolved — by the deadline, or by teardown — while this reply
        // was in flight. Filing it under `deferred` now would register a call
        // that only the re-armed deadline would ever take out again (see
        // deliver()).
        if (call->delivered.load(std::memory_order_acquire)) return;

        QVariant buffered;
        bool haveBuffered = false;
        bool stopping     = false;
        {
            std::lock_guard<std::mutex> g(st->mu);
            stopping = st->stopping.load(std::memory_order_relaxed);
            // The completion can be buffered ALREADY. On one ordered connection
            // it cannot be — the provider writes the sentinel result before the
            // completion event, and both are decoded on the same strand in order
            // — but checking costs one lookup and removes the assumption.
            const auto it = st->completions.find(callId);
            if (it != st->completions.end()) {
                buffered = it->second;
                st->completions.erase(it);
                haveBuffered = true;
            } else if (!stopping) {
                call->callId = callId;      // written under st->mu, read under it
                st->deferred[callId] = call;
            }
        }
        if (haveBuffered) {
            call->deliver(std::move(buffered), logos::CallError{});
            return;
        }
        if (stopping) {
            call->deliver(QVariant(),
                          callErrorReleased(call->objectName, call->method));
            return;
        }
        // A SECOND full deadline, which is what the waiter thread gave it too:
        // it ran the future wait for timeoutMs and then awaitCompletion for
        // another timeoutMs. Preserved deliberately rather than tightened —
        // changing how long a deferred call is allowed to take is a separate
        // decision from removing the thread it used to take it on.
        const int effective =
            call->timeoutMs > 0 ? call->timeoutMs : kDeferredFallbackMs;
        call->armTimer(AsyncCall::clock::now() + std::chrono::milliseconds(effective),
                       effective);
    });
}

bool PlainLogosObject::informModuleToken(const QString& authToken,
                                         const QString& moduleName,
                                         const QString& token,
                                         int /*timeoutMs*/)
{
    EntryGuard guard(this, "informModuleToken()");
    if (!m_conn || !m_conn->isOpen()) return false;
    TokenMessage msg;
    msg.authToken  = authToken.toStdString();
    msg.moduleName = moduleName.toStdString();
    msg.token      = token.toStdString();
    m_conn->sendToken(std::move(msg));
    return true; // fire-and-forget
}

void PlainLogosObject::onEvent(const QString& eventName, EventCallback callback)
{
    EntryGuard guard(this, "onEvent()");
    if (!m_conn || !m_conn->isOpen() || !callback) return;

    {
        std::lock_guard<std::mutex> g(m_mu);
        m_subs.emplace_back(eventName, callback);
    }

    SubscribeMessage msg;
    msg.object    = m_objectName;
    msg.eventName = eventName.toStdString();

    // Bridge RPC event → Qt-flavored callback.
    m_conn->sendSubscribe(std::move(msg), [callback](EventMessage evt) {
        callback(QString::fromStdString(evt.eventName),
                 rpcListToQVariantList(evt.data));
    });
}

void PlainLogosObject::disconnectEvents()
{
    EntryGuard guard(this, "disconnectEvents()");
    disconnectEventsImpl();
}

// The body, without the guard. See the declaration: release() and the destructor
// must not take a reference to an object they are in the middle of destroying.
void PlainLogosObject::disconnectEventsImpl()
{
    std::vector<std::pair<QString, EventCallback>> subs;
    {
        std::lock_guard<std::mutex> g(m_mu);
        subs.swap(m_subs);
    }
    if (!m_conn) return;
    for (const auto& [name, _] : subs) {
        UnsubscribeMessage msg;
        msg.object    = m_objectName;
        msg.eventName = name.toStdString();
        m_conn->sendUnsubscribe(std::move(msg));
    }
}

void PlainLogosObject::emitEvent(const QString& eventName, const QVariantList& data)
{
    EntryGuard guard(this, "emitEvent()");
    if (!m_conn || !m_conn->isOpen()) return;
    EventMessage msg;
    msg.object    = m_objectName;
    msg.eventName = eventName.toStdString();
    msg.data      = qvariantListToRpcList(data);
    m_conn->sendEvent(std::move(msg));
}

QJsonArray PlainLogosObject::getMethods()
{
    EntryGuard guard(this, "getMethods()");
    if (!m_conn || !m_conn->isOpen()) return QJsonArray();

    MethodsMessage msg;
    msg.id     = m_conn->nextId();
    msg.object = m_objectName;

    const std::uint64_t callNumber = msg.id;
    auto fut = m_conn->sendMethods(std::move(msg));
    if (fut.wait_for(std::chrono::seconds(5)) != std::future_status::ready) {
        // Same orphan as the sync call path, on the map next door.
        m_conn->cancelPending(callNumber);
        return QJsonArray();
    }
    auto res = fut.get();
    if (!res.ok) return QJsonArray();
    return methodsToJsonArray(res.methods);
}

void PlainLogosObject::release()
{
    // The RpcConnection is SHARED across every PlainLogosObject a single
    // PlainTransportConnection hands out. Stopping it here would kill
    // the connection for every other holder too, so just unsubscribe our
    // own events and drop our reference — the connection stays alive
    // until PlainTransportConnection itself is destroyed.
    //
    // stopAndCancelCalls() before delete, and it BLOCKS ON NOTHING. In-flight
    // calls used to be threads that captured `this`, so teardown had to ask them
    // to stop and then join them — a wait slice at best, the rest of the call's
    // timeout before that, and a use-after-free if they were merely detached.
    // Now they are shared-owned state that no longer refers to this object at
    // all, so cancelling is a flag, a sweep of the in-flight map, and one
    // callback per abandoned call.
    //
    // That also makes release() safe to call from inside an event callback
    // running on the single io thread — the reentrant shape remote_transport.cpp
    // documents — which a "post a no-op onto the strand and wait for it" barrier
    // could not have been: it would have deadlocked against itself.
    //
    // ── AND A CALL RUNNING ON ANOTHER THREAD RIGHT NOW ──────────────────────
    //
    // That used to be a use-after-free, and it is now safe. The defect was
    // reproduced deterministically on master and on both pending PRs: a
    // synchronous callMethod parked in its future wait, released from a second
    // thread, faults on the very next line it executes
    // (`m_conn->cancelPending(...)`), with and without Guard Malloc, because
    // release() ended in an unconditional `delete this`.
    //
    // WHAT MAKES IT SAFE. The object carries a live-reference count (m_liveRefs,
    // see EntryGuard above): 1 for the owner, plus one per caller currently
    // inside a public entry point. release() does the teardown and then drops
    // THE OWNER'S reference instead of deleting; whoever drops the count to zero
    // does the delete. With a call in flight that is the call's own thread, on
    // its way out, after its last access to the object.
    //
    // AND THE ARGUMENT THAT SAID THIS WAS IMPOSSIBLE, because it is instructive
    // and it was wrong. It ran: any mechanism that could save the racing call has
    // to be reached THROUGH `this`, so the racing thread's first act would be to
    // read a freed object. True — for a call that ENTERS after destruction. It
    // conflates that with a call ALREADY INSIDE the object, which is the defect
    // actually reproduced: that call's reference was taken while the object was
    // provably alive (it is what its own thread did on the way in, before
    // release() was ever called), so its increment is ordered before release()'s
    // decrement in the modification order of m_liveRefs, and release() therefore
    // SEES it. Nothing is read through a dangling pointer anywhere in that
    // sequence. The counter is a member, and that is fine, because the thread
    // that has to trust the member is the thread that already published to it.
    //
    // WHAT IS STILL NOT SAFE, exactly:
    //
    //   * A CALL THAT ENTERS AT OR AFTER release(). Its very first act is to
    //     increment a counter that may already be freed. This is the shape the
    //     old argument describes correctly, it is what "the handle must not be
    //     used after release()" has always meant, and it stays a caller error.
    //     EntryGuard reports it (reportEntryAfterRelease) in the cases where the
    //     object is still alive to notice — i.e. when another call in flight is
    //     holding it — and in the cases where it is not, there is nothing left to
    //     look at.
    //   * `delete obj` INSTEAD OF release() with calls in flight. The caller has
    //     demanded the storage back now, so there is no destruction left to
    //     defer. The destructor reports it (reportConcurrentCallers) and aborts
    //     in debug builds.
    //   * TWO release()es ON THE SAME HANDLE. The second one is a call entering
    //     after destruction, by the definition above.
    //
    // THE COSTS, since deferring a destruction is not free:
    //
    //   * release() still returns immediately, and still waits for nothing — the
    //     property the two preceding changes exist to establish is untouched, and
    //     the reentrant release()-from-an-io-thread-event-callback shape still
    //     cannot deadlock, because nothing here blocks.
    //   * The OBJECT, and with it its share of the connection, now lives until
    //     the last call in flight leaves — bounded by that call's own timeout,
    //     which the caller chose. A handle released while a 30-second call is
    //     parked keeps ~200 bytes and one shared_ptr count alive for the rest of
    //     that call. stopAndCancelCalls() has already cancelled everything it can
    //     wake, so the only thing that can still take the full time is a
    //     synchronous call parked in the connection's future, which nothing in
    //     this transport can interrupt.
    //   * m_conn IS NOT RESET HERE any more, and that is a correctness
    //     requirement rather than a simplification: the parked caller's next act
    //     is `m_conn->cancelPending(...)`, and resetting a shared_ptr while
    //     another thread reads it is a data race on the shared_ptr itself. The
    //     member is released by the destructor, which now runs when nobody is
    //     inside the object.
    m_released.store(true, std::memory_order_release);

    // Impl, not the guarded entry point: an EntryGuard here would take a
    // reference and drop it again, and it would ALSO trip the "entered after
    // release" report it just armed. Same reason in the destructor.
    disconnectEventsImpl();
    stopAndCancelCalls();
    // THE LAST THING THIS FUNCTION TOUCHES, exactly as in EntryGuard: after the
    // decrement this object may belong to another thread, and if the decrement
    // reached zero it does not exist. Nothing below it, ever.
    if (m_liveRefs.fetch_sub(1, std::memory_order_acq_rel) == 1)
        delete this;
}

quintptr PlainLogosObject::id() const
{
    return reinterpret_cast<quintptr>(m_conn.get());
}

} // namespace logos::plain