logos-co/logos-protocol
1
0
Fork 0
mirror of https://github.com/logos-co/logos-protocol.git synced 2026-08-27 12:01:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(protocol): report the failures that happen AFTER acquire — on both twins, without moving the ABI (#41)

Browse Source 
* fix(lp): lp_invoke_async can finally report a failure

lp_result_cb has always been documented as carrying an outcome —
"ok != 0 -> `json` is the result JSON value; ok == 0 -> `json` is the
canonical error object" — and the synchronous twin lp_invoke has always
honoured it (LP_ERR_UNAVAILABLE + out_error_json). lp_invoke_async did
not: it subscribed with the VALUE-ONLY invokeRemoteMethodAsync overload
and called back `cb(1, json, user_data)` with ok hard-coded to 1, so a
call to a module that cannot be acquired reached the callback as a
SUCCESS carrying a default-constructed value.

The fix is to pass a TWO-argument lambda, which is invocable only as
LogosAPIClient::AsyncResultErrorCallback and so binds to the
CallError-aware overload that already exists next to the value-only one.
The failure is then rendered with the same makeErrorJson the sync path
uses, so both entry points report the same event in the same shape.

The ABI is unchanged. lp_result_cb's (ok, json, user_data) signature
already reserves ok == 0 for exactly this; nothing had to grow a new
entry point, and every in-tree consumer already branches on `ok`
(logos-rust-sdk's async_call_trampoline even parses `message` out of the
canonical error object — code written against a contract the
implementation never kept).

Regression test: a matched pair over a REAL transport (plain TCP), not
the mock.

  FAILING async call    -> ok=0 {"code":"object_unavailable", ...}
  SUCCEEDING async call -> ok=1 7

The first fails on the unfixed tree (ok=1, json "null"); the second
passes on both, so an over-eager "report failure everywhere" fix cannot
sneak through.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(protocol): report the failures that happen AFTER acquire, on both twins

#40 made lp_invoke_async able to report a failure, but only for the two
conditions produced ABOVE the transport: acquire failure and the unauthorized
sentinel. Everything the transport learns while the call is in flight was still
discarded — PlainLogosObject answered a bare QVariant() for a timeout and for
`ResultMessage.ok == false` alike, and LogosAPIConsumer hard-coded an empty
CallError next to it.

Two ordinary failures therefore still reported success on both entry points:
a TIMEOUT, and MODULE NOT LOADED against a host that is up (which is not an
acquire failure on the plain wire — requestObject hands back a handle for any
name over an open connection).

The information already exists: ResultMessage carries err/errCode, the futures
know they expired, QtRO knows its pending call never finished. It had nowhere to
go because LogosObject's callMethod returns a lone QVariant and its
callMethodAsync callback takes a lone QVariant.

Widening those virtuals would append a vtable slot to an installed, subclassed
interface, so instead this adds LogosObjectErrorChannel — a SIBLING interface
reached by dynamic_cast. LogosObject's size, layout and vtable are unchanged
(verified: a subclass compiled against the old and new headers emits the same
14-entry vtable with identical slot indices), and a transport that does not
implement it keeps today's behaviour.

logos_protocol.cpp needs no change: lp_invoke and lp_invoke_async already render
this CallError, so both twins gain the coverage together.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(protocol): stop the macOS flake that was sinking #41

Three real races the new CallErrorAfterAcquire suite exposed (and that
Copilot flagged on the QtRO half):

1. ~PlainTransportHost stopped the acceptor but did not quiesce the shared
   Asio I/O thread. Server-side RpcConnections hold a raw IncomingCallHandler*
   back to the host; a fail()/onConnectionClosed racing teardown freed the
   handler mid-call. That is the macOS CI SIGSEGV in
   AsyncSuccessStillReportsTheValue — it fires with no output of its own
   because the previous live-host test's destructor left the heap corrupted.
   Restore the I/O barrier that landed on the qtfree branches but never on
   master (proven: 80/80 clean on the CI crash sequence that was ~2/50 before).

2. PlainLogosObject::callMethodAsync detached its per-call waiter while
   capturing `this`. release()/delete this could then race the waiter.
   Join waiters in the destructor/release, and register the thread under the
   lock before it can outrun teardown.

3. QtRO async could deliver the user callback twice when the timeout timer
   and the pending-call watcher finished around the same moment, violating
   the exactly-once contract. Gate both paths (and the deferred-completion
   arm) on one atomic.

Also drain queued onCall invokes after host.reset() in the #40 live-target
control, matching LiveHost's teardown discipline.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(protocol): the drain barrier must not dangle on its own timeout

Two defects in the barrier added by 1e9c934, both on the path it takes when
it fails:

  std::promise<void> drained;                                   // stack local
  boost::asio::post(ioc, [&drained] { drained.set_value(); });  // by REFERENCE
  fut.wait_for(std::chrono::seconds(5));                        // result dropped

1. The wait is bounded, so on timeout this frame returns while the posted task
   is still queued -- and the task holds a pointer to a destroyed stack object.
   set_value() then writes to freed stack memory. The bound that stops a wedged
   I/O thread hanging teardown introduced the exact class of use-after-free the
   barrier exists to prevent. The promise is now a shared_ptr captured BY VALUE,
   so the task keeps it alive whether or not anyone is still waiting.

2. The wait_for result was discarded. A timeout means the barrier did NOT hold
   and we are about to free an IncomingCallHandler that a live connection may
   still call back into -- the original crash, minus any way to know it
   happened. It now warns, naming the consequence.

Neither is reachable while the I/O thread drains promptly, which is why the
suite is green either way; both matter precisely when it does not, which is
the only situation the barrier is for.

Tests: 270/270.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(protocol): teardown must not wait out the call it is abandoning

Joining the per-call waiters (rather than detaching them) closed a real
use-after-free: the waiter captures `this`, and release() used to `delete this`
underneath it. But joinWaiters() could only join. It had no way to ASK a waiter
to stop, so destroying a PlainLogosObject with a call in flight blocked for the
remainder of that call's timeout — up to 20s on the protocol default. A module
unloading mid-call stalled the unload for that long, on the releasing thread.

Measured, 8s call timeout, provider parked:

    release()                     before        after
    future wait  (site 1)         7804 ms       11 ms
    deferred completion (site 2)  7703 ms        0 ms

Both blocking sites are now interruptible, and they need different treatment:

  * the std::future wait cannot be interrupted at all, so it is SLICED: one
    deadline computed up front, waited in 25ms increments, stop flag checked
    between them. Teardown latency is one slice; the timeout the caller asked
    for is unchanged, because the last slice ends exactly on the deadline. 25ms
    is under two frames (so a module unload stays imperceptible) and costs 40
    wakeups/second per in-flight call, which is nothing beside the Qt event loop
    these threads already sit next to.

  * awaitCompletion's condition_variable is interruptible by construction:
    widen the predicate, notify_all. No latency floor at all — hence 0 ms. The
    flag is published under m_completionMu so a waiter cannot evaluate the
    predicate, decide to sleep, and then miss the notify.

A CANCELLED CALL STILL DELIVERS, EXACTLY ONCE. This is the part a naive fix
breaks: callMethodAsyncWithError and lp_invoke_async promise the callback fires
exactly once, so a waiter that simply returns on stop trades a bounded stall for
an unbounded hang in every caller awaiting it. Proven by building that naive
variant: it passes the latency test and fails three exactly-once tests with the
callback never arriving.

The code is "transport_error", from the existing vocabulary rather than a new
one, since these codes are the wire contract. logos_call_error.h defines it as
"the connection failed or was torn down mid-call", which is precisely what
happened — the consumer tore its own end down. The alternatives all misattribute
it: "object_unavailable" says the module is absent (it is not, and callers
re-acquire on that code), "call_failed" blames the peer for a dispatch it
performed fine, and "timeout" — what this used to report, after waiting the
deadline out — claims a deadline elapsed that did not. It is also already what
the wire produces for the same event seen from the other side (callErrorFromWire
maps TRANSPORT_CLOSED to transport_error).

Delivering during teardown is safe because postToQtEventLoop touches nothing
owned by the object: it is a free function taking the callback, value and error
BY VALUE, and the waiter copies objectName/method up front. That was already
true and is now load-bearing, so it is documented at the function. The queued
lambda runs after the object may be gone; everything the waiter reaches through
`this` runs before the join returns, which is why the join must stay.

Also closes the registration window it opens: a call arriving after the stop
would push a thread onto an m_waiters that teardown has already swapped out, so
it would never be joined. It is answered as cancelled instead.

The UAF is verified still closed under macOS Guard Malloc rather than ASan —
libclang_rt livelocks in its own initializer before main on this toolchain, for
both ASan and TSan, on a hello-world. Under Guard Malloc the race test is clean
across 5 runs and SIGSEGVs immediately when the join is turned back into a
detach, so the check is a real detector and not a vacuous pass.

Tests: 277/277 (was 270; 7 new). CallErrorAfterAcquireTest hammered 40x, 0
failures — it was ~2/50 flaky before this branch's earlier fixes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(protocol): a call that FINISHED must not park its thread for the object's life

The per-call waiters are joinable rather than detached, which is what closed the
use-after-free where release() deleted the object under a still-running waiter
(4f9d824), and they are interruptible, so teardown no longer waits out the call's
timeout (731e579). Both stay. What neither did was retire a waiter that had
FINISHED: m_waiters was only ever swap()ped, in stopAndJoinWaiters(), so an
exited-but-unjoined std::thread — whose stack and pthread struct are not
reclaimed until somebody joins it — stayed parked for the lifetime of the handle.

Measured against a live PlainTransportHost over TCP, every call completing
normally, one handle held throughout, before:

    10000 calls   m_waiters   300 -> 10300   rss +156.56 MiB   16417 B/call
    30000 calls   m_waiters   300 -> 30300   rss +469.28 MiB   16403 B/call

and the same through the production C ABI — one lp_client, N lp_invoke_async —
at +156.53 MiB. That path is why this matters: LogosAPIConsumer caches ONE
handle per module and reuses it for every async call, releasing it only on
eviction or teardown (cpp/logos_api_consumer.cpp:129 and :207), so a
long-lived module leaks per lp_invoke_async. The ~16KB constant is one page on
this 16KiB-page arm64 and will be smaller elsewhere; the UNBOUNDEDNESS is the
platform-independent part, and follows from m_waiters.size() rising 1:1 with
completed calls and only ever falling in teardown. Attribution: the retention
arrived with the join in 4f9d824, not with 731e579 — but 731e579 is what makes
the join permanent.

The registry is now KEYED, because a thread cannot join itself and so a waiter
can never retire its own entry. Each waiter publishes its id as its FINAL act (a
scope guard declared first, so it destructs last, covering all four exit paths),
and the next spawn — plus teardown — joins those ids and erases them. Joining a
thread that has already returned is a couple of syscalls. Same probe, same
workload, after:

    10000 calls   m_waiters    15 -> 16      rss +0.08 MiB         8 B/call
    30000 calls   m_waiters    16 -> 16      rss +0.06 MiB         2 B/call
    10000 calls via lp_invoke_async          rss +0.09 MiB        10 B/call

Retention is now bounded by the waiters that finish after the LAST spawn, i.e.
by peak in-flight concurrency — 16 at the in-flight window above, and exactly 1
when calls are issued sequentially — instead of by call count.

THE DEADLOCK THIS SHAPE INVITES is a reaper that joins while holding m_waiterMu,
against a waiter blocked on m_waiterMu trying to publish. It is avoided by
construction rather than by argument: nothing is joined with a lock held, in the
reaper or in teardown, whatever a waiter does on its way out. Proven by building
the naive variant that does join under the lock — the new hammer wedges it, with
the main thread in reapFinishedWaiters -> pthread_join and a waiter in
publishFinishedWaiter -> mutex wait, and the test's watchdog names the cause
instead of letting CI hang.

Teardown's guarantee is restated rather than weakened. It is not "every waiter
has been joined by the time stopAndJoinWaiters() returns" — a waiter a
concurrent reaper is mid-join on is no longer in the map — but the thing that
guarantee was ever for: NO WAITER TOUCHES THE OBJECT AFTER IT RETURNS. An entry
leaves m_waiters only once its thread has published, and publishing is that
thread's last access.

The TODO above the waiter still stands: the real fix is to fold the wait into
the shared Asio io_context and have no thread per pending RPC at all. This makes
the interim honest; it does not replace that.

Two more things review turned up, folded in here:

  * The two wait sites resolved stop-vs-result in OPPOSITE directions.
    waitForResult tested the stop flag BEFORE polling, so an already-ready
    future was still reported as transport_error, while awaitCompletion
    deliberately preferred a completion that had landed — and both were
    commented as intentional. One rule now, applied to both: AN ANSWER ALREADY
    IN HAND BEATS A CONCURRENT STOP, and the stop only decides what happens when
    there is nothing to hand over. The callback fires either way
    (postToQtEventLoop copies everything it delivers), so the only thing a stop
    can change is what the callback SAYS — and manufacturing transport_error
    while the true answer sits in the future reports a failure that did not
    happen, to callers that re-acquire, retry and log on that code. Preferring
    the answer costs nothing, since it is already there: the flag is still
    checked before every sleep, so the teardown-latency bound is unchanged.

  * CORRECTION to 731e579's message, which claimed it "closes the registration
    window" where a call arriving after the stop would never be joined. That
    branch is unreachable in defined behaviour: m_stopping is raised only by
    teardown, so any thread that can read it inside callMethodAsyncWithError is
    already calling a method on an object whose destructor is running — the load
    is itself the use-after-free, reproduced as a SIGSEGV on that commit and on
    its parent alike, and nothing inside that function can repair it. The guard
    is harmless and stays (one predictable branch, and it fails safe with one
    callback), but its comment now says what it is instead of claiming a fix it
    does not make.

Verified by running, with every check first shown to FAIL on unfixed code:

  * Retention: the probe above, plus a committed regression test that reads
    m_waiters out of the live object through the explicit-instantiation access
    hole ([temp.spec] does not check access on an explicit instantiation's
    template arguments) — so the code under test keeps its private state, with
    no friend, no test-only accessor and no `#define private public`. 200
    sequential completed calls keep 1 waiter; without pruning they keep 200.
  * Exactly-once on all four paths — normal completion, timeout, cancellation
    and the deferred-completion (pending-sentinel) arm — counted PER CALL so a
    dropped one and a doubled one cannot cancel out, plus the 60-round
    release-during-call race. Shown to catch a cancelled path that returns
    silently (3 failures) rather than delivering.
  * Teardown latency unchanged from 731e579: 10-17ms with an in-flight 8000ms
    call and 0-1ms mid-defer, against 15ms / 1ms on that commit.
  * The UAF stays closed: 11 teardown + reaping tests clean under macOS Guard
    Malloc (ASan/TSan remain unusable on this toolchain).
  * Full suite 281/281 twice, `nix build .#tests` green (281/281 in the
    sandbox), CallErrorAfterAcquireTest hammered 40x clean.

* fix(protocol): a burst that goes quiet must not wait for a call that never comes

378d889 retired finished waiters, but from ONE site: the async-call spawn path.
So whatever finishes after the LAST spawn is never reaped, and a module that
bursts and then goes idle parks it all until the handle dies. Measured on
378d889, one handle, 2000 concurrent calls, every one delivered:

    after 2000 completed calls, IDLE:  m_waiters=1428   rss=+24.17 MiB
    after ONE further call:            m_waiters=1      rss=+ 1.92 MiB

The unbounded-per-call class was gone; this is what it left behind, and the
second line is the whole diagnosis — the corpses go the instant anything calls
again, so the reaper works and simply never runs. LogosAPIConsumer caches one
handle per module and never releases it between calls, so "bursts, then quiet"
is not a corner case: it is a UI that fans out on a refresh and then waits for
the user.

A finishing waiter now reaps the OTHER finished waiters before publishing
itself, so a burst drains as it completes. Same probe, same workload:

    after 2000 completed calls, IDLE:  m_waiters=1      rss=+ 1.88 MiB

THE BOUND IS ONE, NOT ZERO, and by construction rather than by luck: a waiter
can only reap OTHERS (a thread cannot join itself), so the last one to finish
has nobody behind it to collect it. Anything that publishes after the final
reap survives too, which is why 12 runs of the probe gave 1 eleven times and 2
once. Those go on the next call, or in teardown. Retention now tracks neither
call count nor peak concurrency — the sequential and in-flight-16 numbers move
from "15 -> 16 waiters" to "1 -> 1" — and the memory figures are unchanged
against 378d889 where they were already flat: 10k sequential +0.00 MiB, 10k at
16 in flight +0.06 MiB, 30k +0.09 MiB, and 10k through the production C ABI
(one lp_client, N lp_invoke_async) +0.09 MiB / 10 B per call, the same as
378d889 reported.

THE ORDER IS THE SAFETY ARGUMENT. Reap first, publish last, never the reverse:

  * Publishing is what makes a waiter joinable BY ANOTHER WAITER. Reaping first
    keeps that relation one-way — unpublished threads join published ones,
    published ones join nobody — so it has no cycles. Inverted, two waiters
    publishing in the same instant can each take the other's thread out of
    m_waiters and then join it; both are already out of the registry, so
    teardown does not even wait for them. Built that variant: pthread_join
    detects the cycle and throws, the half-drained thread vector then destroys
    a still-joinable thread, and the process aborts — the EXISTING hammer
    (ReapingRacesPublishingWithoutDeadlocking) catches it 5 runs out of 5, with
    the stack showing two waiters inside FinishOnExit joining each other.
  * While a waiter is unpublished it is still in m_waiters, so a concurrent
    teardown joins it and the object cannot be destroyed under the reap. Once
    published, a reaper may take its thread out of the map and release() may
    `delete this` — and a reaper on the CALLER's thread (the spawn path) is one
    teardown neither knows about nor waits for, so a post-publish touch of
    m_waiterMu is a use-after-free on a member mutex. That path needs a caller
    still issuing calls while another thread releases, which this class already
    treats as caller-side UB, so it is stated as an argument; the cycle above is
    what the tests actually demonstrate.

Two corrections to 378d889, which this change makes load-bearing rather than
cosmetic. NOT amended into it — it is pushed, and a commit that misstates its
own reasoning is better read alongside the correction than rewritten.

  * plain_logos_object.h:107-109 said reapFinishedWaiters() is "called on every
    async spawn ... and from stopAndJoinWaiters()". It is not, and never was,
    called from stopAndJoinWaiters(): teardown does its own id-independent
    brute-force join, which is precisely why it needs no cooperation from the
    reaper. Harmless behaviourally, wrong in a mechanism whose entire argument
    is who joins what and when. The comment now names the two real callers —
    the spawn path and, as of this commit, every waiter on its way out.

  * 378d889's message presented "the join is outside the lock" as THE property
    that prevents the reaper deadlock, "proven by construction" by its hammer.
    That is overstated, in a way that would let the guarantee be refactored
    away with the suite still green. TWO independent properties each suffice:
    joining only PUBLISHED ids (a published waiter never needs m_waiterMu
    again, so it cannot be the thread being shut out), and joining outside the
    lock. The hammer only wedges when BOTH are gone. Measured, on top of this
    change: the variant that joins under the lock but KEEPS the published-only
    filter passes ReapingRacesPublishingWithoutDeadlocking in 293/297/290ms
    across three runs and the whole reaping suite besides, while the variant
    that joins everything under the lock trips the watchdog at 60s. So a later
    "simplification" that moves the join inside the lock would ship green. Both
    properties are kept, and the comment now says which one the test is
    actually testing.

The TODO above the waiter still stands: the real fix is to fold the wait into
the shared Asio io_context and have no thread per pending RPC at all. This
makes the interim honest; it does not replace it.

Verified by running, each check first shown to FAIL on unfixed code:

  * Retention: the burst probe above, plus a committed regression test that
    reads m_waiters out of the live object through the explicit-instantiation
    access hole. 800 concurrent completed calls, then IDLE with NO further
    call: 1 waiter left, 20 runs out of 20. On 378d889 the same test leaves
    610 of 800 and fails. The pre-existing sequential and in-flight tests are
    unchanged and still pass.
  * The UAF stays closed — the check that matters most here, because this adds
    an object access late in the waiter's life. 9 reaping/teardown-race tests
    plus the 7-test teardown suite clean under macOS Guard Malloc (ASan is
    unusable on this box: it hangs in its own initializer). DETECTOR VALIDATED
    both ways: turning teardown's join back into a detach SIGSEGVs under Guard
    Malloc on the release-during-call hammer (exit 139), and the specific
    inversion this change risks — reaping AFTER publishing — aborts as
    described above.
  * No deadlock: reap-vs-publish hammered 20x (1600 calls in 40 overlapping
    bursts each), plus 60 rounds of teardown landing from another thread while
    the tail of a burst retires itself, plus 6x600-call bursts checking that
    LIVE OS threads (task_threads, which counts wedges and not corpses) come
    back to baseline every round. Clean; the watchdog names the cause if it
    ever is not.
  * Exactly-once on all four paths — normal, timeout, cancellation, deferred
    sentinel — counted per call. Each detector validated with a broken build:
    dropping the cancelled callback fails 4 tests, dropping the timeout one
    fails its test, and double-delivering the normal/deferred arm fails those.
  * Teardown latency unchanged from 378d889: 1-25ms with an in-flight 8000ms
    call and 0ms mid-defer across 5 runs, against 2-21ms / 0ms on that commit —
    the same one-wait-slice (25ms) bound, since a waiter's extra work happens
    after it has stopped waiting.
  * Full suite 282/282 three times, `nix build .#tests` green,
    CallErrorAfterAcquireTest hammered 40x clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* test(protocol): pin publishing as a waiter's LAST access to the object

PlainLogosObject's waiters are joinable, interruptible and reaped, and all
three rest on one ordering rule that nothing in the suite could see:

    ~FinishOnExit() {
        self->reapFinishedWaiters();      // others, never itself
        self->publishFinishedWaiter(id);  // strictly last
    }

reapFinishedWaiters() erases published entries from m_waiters under m_waiterMu
and joins those threads OUTSIDE it. stopAndJoinWaiters() swaps m_waiters under
the same lock and brute-force joins whatever it got. So a waiter that a
concurrent reaper is mid-join on is NOT in teardown's map, and teardown can
return — with release() going straight on to `delete this` — while that waiter
is still unwinding. stopAndJoinWaiters() already says this in as many words:
the guarantee is not "everything is joined when this returns" but "no waiter
touches this object after this returns". Publishing being last is the entire
reason the second sentence is true, so one member access below it is a
use-after-free, and moving the publish above the reap is a join cycle.

THE DEFECT SHIPS GREEN. Rebuild plain_logos_object.cpp with a single object
read after the publish and the whole of PlainObjectTeardownTest and
PlainWaiterReapingTest passes, 10 runs out of 10, cleanly under Guard Malloc.
That is not a hole in those suites. No SUPPORTED caller can provoke it: under
calls-in-flight-plus-release, every waiter is still joined transitively,
because a waiter leaves m_waiters only via teardown (which joins it) or a
reaper, and a reaper is either another waiter — itself in m_waiters until after
its join returns — or the async-spawn path, whose join completes before the
call returns. The one uncovered reaper is the spawn path racing a concurrent
release(), and calling a method on an object another thread is releasing is
caller-side UB that faults on correct code too. A test built on that race would
be red on green code, so it is not a usable detector.

SO STOP RACING AND OBSERVE. tests/protocol/test_plain_waiter_publish_is_last.cpp
drives a real PlainLogosObject through a scripted RpcConnectionBase — no socket,
no host, no event-loop timing, and the test decides exactly when the call's
future is satisfied — and watches the accesses in two halves.

  * THE STATE. The object is placement-newed into an mmap'd two-page arena, put
    down so a page boundary lands at m_waiterMu: the members teardown
    coordinates on go on the second page, everything else on the first. The
    first page is mprotect(PROT_NONE)'d for exactly as long as a waiter runs,
    and a SIGSEGV/SIGBUS handler RECORDS each access — address, thread, and how
    many ids were published at that instant — then unprotects so the access
    proceeds. Nothing crashes; the access is evidence. A correct waiter touches
    that page zero times: objectName and method are copied into the closure
    precisely so it needs nothing from the object. Four rounds, one per exit
    path out of the lambda (answered, rejected, timed out, cancelled), since all
    four end in the same guard.

  * THE REGISTRY, which that page cannot cover because publishing has to reach
    it. Caught with bait, using the reaper's own shape: reapFinishedWaiters()
    joins outside m_waiterMu, so a waiter that has picked up somebody else's
    finished thread sits in that join holding nothing — a window the test holds
    open as long as it likes, because the thread being joined is one the test
    planted and keeps parked. Plant bait 1; let the call finish; the exit guard
    reaps, takes it, parks. Plant bait 2 at leisure. Release bait 1; the waiter
    finishes its reap and publishes. Bait 2 must still be registered. Bait 1
    doubles as a check that the reap really does join with the lock free.

Neither half is probabilistic. A third test proves the detector can fire at all,
so the two "this counter stayed at zero" assertions are not vacuous.

MEASURED, rebuilding the file under test with each defect (caught/runs):

  defect below publishFinishedWaiter()      new    teardown+reaping
  ------------------------------------      ---    ----------------
  read m_objectName                       40/40                0/10
  read m_conn                             10/10                0/10
  read m_completions                      10/10                0/10
  read m_completionSubscribed             10/10                0/10
  lock m_mu                               10/10                0/10
  call reapFinishedWaiters() again        20/20                 2/2
  read m_stopping                          0/10                0/10
  (publish moved ABOVE the reap)            0/5               12/15
  no defect — 8f0c60f                      0/40                0/10

The one gap is m_stopping, the single member sharing the registry's page, which
cannot be guarded without guarding the publish. The inverted order is left to
the reaping suite's hammer, which has it covered. Runtime 0.9-1.0s for all
three tests; clean 40/40 on 8f0c60f, and clean 3/3 under Guard Malloc
(MALLOC_PROTECT_BEFORE=1, banner confirmed) — the test never touches freed
memory itself, which is the other half of not being built on UB. No Guard
Malloc needed to detect anything: mprotect and the bait are the detectors.

Also: nix build .#tests 100% (285/285), the full binary 285/285, and
CallErrorAfterAcquireTest 40/40.

CORRECTIONS to measurements claimed earlier on this branch. All three were
overstated in the same direction — a single sample read as a constant:

  * "ReapingRacesPublishingWithoutDeadlocking aborts the process, 5 runs out of
    5" (plain_logos_object.cpp, and 378d889's message) is 12 runs in 15, ~80%.
    It is a race detector, so one green run of it proves nothing — which is
    exactly the argument for the deterministic suite added here. Corrected in
    the comment.
  * C-ABI retention was reported as "+0.09 MiB / 10 B per call" for 10k
    lp_invoke_async on one client (8f0c60f's message). ~6 B/call. Same
    conclusion — flat — different arithmetic.
  * The burst retention figures 1428 (2000 calls, idle) and 610 of 800 came
    back as 1421 and 599 on re-measure of the same build. Race-dependent, same
    magnitude, which is why the tests assert a bound and not a value. Noted in
    test_plain_waiter_reaping.cpp so the next reader does not treat them as
    reproducible constants.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* test(protocol): make the publish-is-last probe fail loudly, and state the rule it actually checks

Three defects in cb015f5's regression test, plus the corrections that commit's
own CORRECTIONS section still owed. No behaviour change: the diff against
8f0c60f under cpp/ is comment lines only, verified by filtering the diff.

(a) THE BAIL-OUT PATH HUNG INSTEAD OF FAILING, which is the one that can stall
    CI. PublishedWaiterDoesNotTouchTheRegistryAgain plants parked "bait" threads
    behind gates and registers them in m_waiters. An ASSERT that fires before
    gate1.open() — ASSERT_TRUE(tookBait1) is the obvious one — returns from the
    function, and then ~PlainLogosObject blocks forever joining a thread nobody
    will release. Reproduced by removing the reap from the exit guard: the
    assertion PRINTS and the run still ends as a timeout kill, exit 124, with no
    test result at all.

    The gates are now opened by a scope guard on every exit path, and declared
    BEFORE the GuardedObject so they outlive the teardown that joins the threads
    parked on them. Same break, after: the same assertion, exit 1, 10.0s — which
    is the probe's own tryWithRegistry budget and not a hang.

    This is the shape a future refactor trips, not a hypothetical: the TODO
    above the waiter (fold the wait into the shared Asio io_context) moves where
    reaping happens, which is exactly the edit that makes tookBait1 false.

    Both tests also stopped capturing their delivery counter by reference. On a
    bail-out the cancelled call's callback is delivered on a later event-loop
    iteration, i.e. after the frame is gone — a real use-after-free on the way
    out of a failing test in a file about use-after-free. Owned by the callback
    now.

(b) THE STATE ASSERTION WAS STRICTER THAN THE INVARIANT. It asserted
    accessCount() == 0; the rule is only "no access AFTER the publish", and the
    fault handler already stamps each access with how many ids were published at
    that instant, so it can tell them apart.

    Not hypothetical either. On the deferred/"multi" path a CORRECT waiter calls
    awaitCompletion() (plain_logos_object.cpp:338), which locks m_completionMu
    and reads m_completions and m_objectName — all on the guarded page, all
    before it publishes. cb015f5 was green only because none of its four rounds
    returned a pending sentinel, and the header's claim that "a correct waiter
    touches that page ZERO times, before the publish or after" was true only of
    the non-deferred rounds.

    So: a fifth round drives the pending-sentinel path (ScriptedConn now answers
    with the sentinel; nothing pushes the completion, so awaitCompletion runs out
    its deadline), and the assertion narrowed to accesses stamped published >= 1.
    PROVEN BOTH WAYS on this tree — with the old accessCount() == 0 predicate the
    new round fails on correct code, naming offset 144 with "0 waiter id(s)
    already published"; with the narrowed one the suite is 30/30 clean.

    The round cannot pass vacuously: it REQUIRES at least one recorded access, so
    a machine slow enough to turn it into a plain timeout fails it instead of
    quietly proving nothing. Each round also now asserts m_finishedWaiters is
    empty before arming, which is what makes "published >= 1" mean "after THIS
    waiter's publish".

    Two things guard against the narrowing being a quiet disarm:

      * the handler now re-arms. It could not before (the faulting instruction
        re-runs immediately), so the observing thread does it — it polls the
        registry anyway and never touches the guarded page. Without it the first
        legitimate access disarms the detector for the whole round.
      * the catch rates were re-measured, not assumed. They are unchanged.

(c) TWO OVERSTATED NUMBERS, in the section whose whole point was to stop
    overstating. Fixed where they live; cb015f5 is pushed and is not rewritten.

      * "C-ABI retention ~6 B/call, not 10" replaced one sample with another.
        10k lp_invoke_async on one lp_client, run ten times: 0, 5, 5, 5, 7, 7, 8,
        10, 13, 10 bytes/call (mean 7.0). Ten more, run here: 3, 11, 8, 5, 8, 10,
        13, 8, 3, 10 (mean 7.9). One distribution, range 0-13; both 6 and 10 sit
        inside it and 8f0c60f's arithmetic (0.09 MiB / 10k) was not wrong.
        THE HONEST STATEMENT IS THAT IT IS FLAT: indistinguishable from zero, RSS
        noise and not a per-call rate. Recorded in test_plain_waiter_reaping.cpp
        beside the other retention figures, where the next person to quote one
        will see it.
      * the table cell "reapFinishedWaiters() again ... 2/2" for the older
        teardown+reaping suites was a two-run sample printed beside 10-40 run
        samples. Re-measured over 30 runs: 9/30 here, 12/30 on another 30-run
        sample — roughly one run in three, matching what reapFinishedWaiters'
        own comment already said ("about one run in four"). The cell now reads
        9/30, and the table says to read that column as rates and the left-hand
        one as deterministic.

ALSO STATED PLAINLY, because it was overstated in review: the window where
"teardown returns while a reaped waiter is still unwinding" is NOT reachable by
a supported caller. A waiter leaves m_waiters only via teardown (which joins it)
or via a reaper, and that reaper is either another waiter — still registered
itself, since it reaps before it publishes, so teardown joins it and therefore
waits out the join it is in — or the async-spawn path, whose join completes
before the call returns. The only uncovered reaper is the spawn path racing a
concurrent release(), which is caller-side UB on any version of this class.

So publish-is-last is an invariant the design rests on and documents, not a
lurking use-after-free. This suite pins it against future edits; it does not
close an open hole. The file header, both failure messages and the comment in
plain_logos_object.cpp now say that instead of implying otherwise.

MEASURED AFTER THE CHANGE, rebuilding plain_logos_object.cpp with each defect
below publishFinishedWaiter() and running the suite (caught/runs), beside the
numbers from before it:

  defect                                    before      after
  ------                                    ------      -----
  read m_objectName                          25/25      25/25
  lock m_mu                                  25/25      25/25
  write m_completions under m_completionMu   25/25      25/25
  call reapFinishedWaiters() again (bait)    20/20      20/20
  read m_conn                                    -      10/10
  read m_completions                             -      10/10
  read m_completionSubscribed                    -      10/10
  read m_stopping (declared blind spot)       0/10       0/10
  publish moved ABOVE the reap (delegated)     0/5        0/5
  no defect                                   0/30       0/30

Nothing moved, including the two declared blind spots — a narrowing that had
started catching or stopped catching something would show here. cb015f5 reported
40/40 for m_objectName from a longer run; 25/25 is this run, not a regression.

WHERE THE DEFERRED ROUND IS WEAKER, said here rather than left to be found: on
that one round the post-publish half is best-effort. A legitimate access opens
the page, the re-arm is a syscall behind, and a defect firing a microsecond
later slips through — measured with every round forced to run, the other four
catch a post-publish m_objectName read 5/5 and the deferred round 0/5, and a
variant that spins on the re-arm instead of polling records 4-24 accesses per
round and still catches it 0/5. It costs nothing: FinishOnExit is ONE piece of
code shared by all five exit paths, so the same defect is the same defect on
every round and the other four catch it deterministically. The deferred round is
there to keep the assertion honest about correct code, not to add a fifth copy
of the same detection.

Verified: PlainWaiterPublishIsLastTest 30/30 clean, the three waiter suites
15/15, the full binary 285/285, `nix build .#tests` 100% (285/285),
CallErrorAfterAcquireTest 40/40. Suite runtime 1.3-1.5s for the three tests
(0.9-1.0s before — the deferred round waits out a 400ms completion deadline).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Dario Lipicar
2026-08-06 12:29:53 -03:00
committed by GitHub
co-authored by Claude Opus 5 Cursor
parent d0523c1486
commit 0f26ffdeef
16 changed files with 3889 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/implementations/plain/plain_logos_object.cpp
+473 -31
View File
@@ -9,13 +9,114 @@
#include <QTimer>
#include <QVariantMap>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <future>
#include <string>
#include <thread>
#include <utility>
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 one rule both waits follow ──────────────────────────────────────────
//
// AN ANSWER ALREADY IN HAND BEATS A CONCURRENT STOP. The stop only decides what
// happens when there is nothing to hand over.
//
// The two sites used to resolve this in opposite directions — this one tested
// the flag before polling, so an already-ready future was still reported as
// transport_error, while awaitCompletion deliberately preferred a completion
// that had landed. Both were commented as deliberate, and they cannot both be
// right, so: the callback fires either way (postToQtEventLoop copies everything
// it delivers, precisely so a released handle costs it nothing), which means the
// only thing a stop can change is what the callback SAYS. Reporting
// transport_error while the true answer sits in the future is a failure that did
// not happen, and that code is not inert — callers re-acquire, retry and log on
// it. Preferring the answer is also free: it is already there, so nothing waits
// for it. The teardown-latency bound is untouched, because the flag is still
// checked before every sleep, and a stop with no answer in hand still wins
// immediately.
//
// 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<ResultMessage>& fut, int timeoutMs,
const std::atomic<bool>& stopping)
{
using clock = std::chrono::steady_clock;
const auto deadline = clock::now() + std::chrono::milliseconds(timeoutMs);
for (;;) {
// Poll FIRST, with a zero wait: an answer in hand beats both a
// concurrent stop and the deadline. This is also what makes a
// non-positive timeout behave as the single unsliced wait_for did —
// one poll, then give up — and what reports a future that went ready
// during the last slice.
if (fut.wait_for(clock::duration::zero()) == std::future_status::ready)
return WaitOutcome::Ready;
// 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.
if (stopping.load(std::memory_order_acquire))
return WaitOutcome::Cancelled;
const auto remaining = deadline - clock::now();
if (remaining <= clock::duration::zero())
return WaitOutcome::TimedOut;
fut.wait_for(std::min<clock::duration>(kWaitSlice, remaining));
}
}
// 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<RpcConnectionBase> conn)
: m_objectName(std::move(objectName))
@@ -26,6 +127,119 @@ PlainLogosObject::PlainLogosObject(std::string objectName,
PlainLogosObject::~PlainLogosObject()
{
disconnectEvents();
stopAndJoinWaiters();
}
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<std::mutex> g(m_completionMu);
m_stopping.store(true, std::memory_order_release);
}
m_completionCv.notify_all();
}
void PlainLogosObject::publishFinishedWaiter(std::uint64_t id)
{
std::lock_guard<std::mutex> g(m_waiterMu);
m_finishedWaiters.push_back(id);
}
void PlainLogosObject::reapFinishedWaiters()
{
// Only ids a waiter published are taken, and publishing is that waiter's
// last act — so everything moved into `done` has already stopped touching
// this object, and joining it is effectively instant.
std::vector<std::thread> done;
{
std::lock_guard<std::mutex> g(m_waiterMu);
std::vector<std::uint64_t> keep;
for (const std::uint64_t id : m_finishedWaiters) {
const auto it = m_waiters.find(id);
if (it == m_waiters.end())
continue; // teardown already took this one
if (it->second.get_id() == std::this_thread::get_id()) {
// A waiter DOES run this now, on its way out — but always
// BEFORE it publishes, so its own id cannot be in the list it
// is walking, and this branch stays unreachable. It costs one
// comparison, and it turns the ordering slip that would make it
// reachable (publishing before reaping) into a leaked entry
// rather than a self-join, which throws out of the noexcept
// destructor doing the reaping and takes the process with it.
// Leave it registered; the next reaper, or teardown, collects it.
keep.push_back(id);
continue;
}
done.push_back(std::move(it->second));
m_waiters.erase(it);
}
m_finishedWaiters.swap(keep);
}
// Joined with NO lock held. The deadlock this whole mechanism can
// introduce is a reaper that holds m_waiterMu while it joins a waiter which
// is itself blocked on m_waiterMu trying to publish. TWO INDEPENDENT
// PROPERTIES each prevent it, and either one alone would be enough:
//
// * only PUBLISHED ids are joined, and publishing is a waiter's last
// access — so a thread this function joins can never be a thread that
// still wants m_waiterMu;
// * no join happens with a lock held, so even joining a thread that DID
// still want the mutex could not shut it out.
//
// Both are kept on purpose: the filter is a property of the logic here,
// which a refactor can lose without looking wrong, while "no join under a
// lock" is structural and tends to survive one. Be precise about what that
// costs in testing, though — the hammer in the regression suite only wedges
// when BOTH are gone. A variant that joins under the lock but keeps the
// published-only filter passes it, measured, in the usual few hundred ms.
// stopAndJoinWaiters() keeps the same discipline.
for (auto& t : done) {
if (t.joinable())
t.join();
}
}
void PlainLogosObject::stopAndJoinWaiters()
{
stopWaiters();
std::map<std::uint64_t, std::thread> waiters;
{
std::lock_guard<std::mutex> g(m_waiterMu);
waiters.swap(m_waiters);
}
// Joined with NO lock held: a waiter on its way out still takes
// m_completionMu (awaitCompletion) and then m_waiterMu (to publish), and
// m_waiterMu is also what a concurrent callMethodAsyncWithError needs in
// order to see the stop flag.
//
// Everything outstanding is joined by id-independent brute force, so this
// needs no cooperation from the reaper: a waiter that publishes while this
// loop runs simply leaves a stale id behind, and its thread is joined here
// anyway.
//
// A waiter that a concurrent reaper is in the middle of joining is NOT in
// this map, and that is still safe. The invariant is not "every waiter has
// been joined by the time this returns" but the thing that invariant was
// ever for: NO WAITER TOUCHES THIS OBJECT AFTER THIS RETURNS. An entry
// leaves m_waiters only once its thread has published, and publishing is
// that thread's last access — all it has left to do is unwind.
for (auto& entry : waiters) {
std::thread& t = entry.second;
if (t.joinable())
t.join();
}
// Cleared after the joins, so the stale ids just described go too. Nothing
// can be added afterwards: m_stopping is set, so no new waiter registers.
{
std::lock_guard<std::mutex> g(m_waiterMu);
m_finishedWaiters.clear();
}
}
QVariant PlainLogosObject::callMethod(const QString& authToken,
@@ -33,7 +247,24 @@ QVariant PlainLogosObject::callMethod(const QString& authToken,
const QVariantList& args,
int timeoutMs)
{
if (!m_conn || !m_conn->isOpen()) return QVariant();
// 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)
{
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).
@@ -50,12 +281,22 @@ QVariant PlainLogosObject::callMethod(const QString& authToken,
if (fut.wait_for(std::chrono::milliseconds(timeoutMs)) != std::future_status::ready) {
qWarning() << "PlainLogosObject::callMethod: timeout for" << methodName;
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);
@@ -64,7 +305,7 @@ QVariant PlainLogosObject::callMethod(const QString& authToken,
{
QString callId;
if (logos::isPendingCallSentinel(value, &callId))
return awaitCompletion(callId, timeoutMs);
return awaitCompletion(callId, timeoutMs, methodName, err);
}
return value;
}
@@ -90,20 +331,43 @@ void PlainLogosObject::ensureCompletionSub()
});
}
QVariant PlainLogosObject::awaitCompletion(const QString& callId, int timeoutMs)
QVariant PlainLogosObject::awaitCompletion(const QString& callId, int timeoutMs,
const QString& methodName,
logos::CallError* err)
{
std::unique_lock<std::mutex> lk(m_completionMu);
const auto effectiveMs = timeoutMs > 0 ? timeoutMs : 30000;
const auto deadline = std::chrono::steady_clock::now()
+ std::chrono::milliseconds(timeoutMs > 0 ? timeoutMs : 30000);
const bool got = m_completionCv.wait_until(lk, deadline,
[&] { return m_completions.count(callId) > 0; });
if (!got) {
qWarning() << "PlainLogosObject: deferred call" << callId << "timed out";
+ std::chrono::milliseconds(effectiveMs);
// 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 — the same rule
// the future wait follows (see waitForResult): 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 = 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 {
@@ -118,14 +382,23 @@ 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.
void postToQtEventLoop(PlainLogosObject::AsyncResultCallback callback,
QVariant result)
//
// 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)
{
QCoreApplication* app = QCoreApplication::instance();
if (!app) return;
QMetaObject::invokeMethod(app,
[callback = std::move(callback), result = std::move(result)]() mutable {
callback(result);
[callback = std::move(callback), result = std::move(result),
err = std::move(err)]() mutable {
callback(result, err);
},
Qt::QueuedConnection);
}
@@ -137,12 +410,30 @@ void PlainLogosObject::callMethodAsync(const QString& authToken,
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)
{
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.
postToQtEventLoop(std::move(callback), QVariant());
postToQtEventLoop(std::move(callback), QVariant(),
logos::callErrorTransport(
m_objectName,
"connection to '" + m_objectName + "' is not open"));
return;
}
@@ -163,23 +454,165 @@ void PlainLogosObject::callMethodAsync(const QString& authToken,
// future iteration can fold this wait into the shared Asio
// io_context (the connection already runs on it) so we don't spin
// up a thread per pending RPC.
std::thread([this, fut, timeoutMs, callback = std::move(callback)]() mutable {
if (fut->wait_for(std::chrono::milliseconds(timeoutMs))
!= std::future_status::ready) {
postToQtEventLoop(std::move(callback), QVariant());
//
// The thread is JOINed — in reapFinishedWaiters() once it has finished, or
// in stopAndJoinWaiters() (destructor / release) if teardown gets there
// first — never 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;
const std::string method = methodName.toStdString();
// Retire the previous calls' waiters before adding one. The waiters reap
// each other too, on their way out (see the guard below) — that is what
// drains a burst which then goes quiet, and it is why this site is no
// longer the only reaper. It still earns its keep: a waiter can only reap
// OTHERS, so the last one to finish has nobody behind it to collect it.
// Done BEFORE taking m_waiterMu because it joins, and joining under that
// lock is the shape described in reapFinishedWaiters().
reapFinishedWaiters();
// Register under the lock BEFORE the thread can outrun release(): a
// detach-then-push left a window where delete this raced the waiter.
{
std::lock_guard<std::mutex> g(m_waiterMu);
if (m_stopping.load(std::memory_order_acquire)) {
// Refuse rather than register: teardown has already swapped
// m_waiters out, so a thread pushed now would never be joined.
//
// To be honest about what this branch is: it is NOT a reachable
// window that got closed. m_stopping is raised only by teardown
// (release() / the destructor), so a thread 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 inside this function can repair that. Reproduced as a
// SIGSEGV, on this branch and on its parent alike. It is kept
// because it costs one predictable branch on a path that already
// does a socket write, and because failing this way — one callback,
// with the same error a cancelled call gets — is strictly better
// than pushing a thread nobody will ever join, should some future
// caller of stopWaiters() make the state legitimately observable.
postToQtEventLoop(std::move(callback), QVariant(),
callErrorReleased(objectName, method));
return;
}
auto res = fut->get();
QVariant value = res.ok ? rpcValueToQVariant(res.value) : QVariant();
// Resolve a "multi" provider's deferred completion (sentinel → wait for
// the completion event) right here on the waiter thread.
{
QString callId;
if (logos::isPendingCallSentinel(value, &callId))
value = awaitCompletion(callId, timeoutMs);
}
postToQtEventLoop(std::move(callback), std::move(value));
}).detach();
const std::uint64_t waiterId = m_nextWaiterId++;
std::thread waiter([this, waiterId, objectName, fut, timeoutMs, methodName, method,
callback = std::move(callback)]() mutable {
// Everything reached through `this` below (m_stopping,
// awaitCompletion's m_completionMu / m_completions) is safe only
// because this thread is joined before the object dies — by the
// reaper if it finishes first, by stopAndJoinWaiters() otherwise.
// 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.
//
// Declared FIRST so it destructs LAST: publishing this waiter's id
// is what permits somebody else to join and drop it, so it must
// come after every access to the object, on every exit path
// (four returns below, plus anything that throws). What runs after
// it is the unwinding of the captures above, none of which belongs
// to the object: a string, a shared_ptr to the call's future, and a
// callback that has already been moved out.
//
// It REAPS BEFORE IT PUBLISHES, and that order is the safety
// argument rather than a stylistic choice. Two reasons, one of
// which the test suite demonstrates:
//
// * Publishing is what makes a waiter joinable BY ANOTHER WAITER.
// Reaping first keeps that relation one-way — unpublished
// threads join published ones, published ones join nobody — so
// it cannot contain a cycle. Inverted, two waiters that publish
// in the same instant can each take the other's thread out of
// m_waiters and then join it. Both are already out of the
// registry, so teardown does not even wait for them; here
// pthread_join detects the cycle and throws out of
// reapFinishedWaiters, whose half-drained vector then destroys
// a still-joinable thread — std::terminate. Re-measured over a
// longer run than the 5/5 an earlier commit message claimed:
// with the two lines below swapped, ReapingRacesPublishingWith-
// outDeadlocking aborts the process 12 runs in 15. It is a
// race, so it is a probabilistic detector and a single green
// run of it proves nothing.
// * Until it publishes, this waiter is still in m_waiters, so a
// concurrent teardown joins it and the object cannot be
// destroyed under the reap. After publishing, a reaper can take
// its thread out of the map and release() can `delete this`,
// and a reaper running on the CALLER's thread (the spawn path
// above) is one teardown neither knows about nor waits for — so
// the touch of m_waiterMu would land on freed memory. That one
// needs a caller still issuing calls while another thread
// releases, which this class already treats as caller-side UB,
// so no test can provoke it without being red on correct code.
// BE PRECISE ABOUT WHAT THAT MAKES THIS: an invariant the design
// rests on, NOT a live use-after-free waiting to be hit. Under
// supported use every waiter is still joined transitively — one
// leaves m_waiters only via teardown (which joins it) or via a
// reaper, and that reaper is either another waiter, still
// registered itself because it reaps before it publishes, or the
// spawn path, whose join finishes before the call returns. The
// only reaper nobody waits for is that spawn path racing a
// release(), i.e. the caller-side UB above.
// test_plain_waiter_publish_is_last.cpp therefore stops trying
// to provoke it and OBSERVES the accesses instead: it guards the
// object's non-registry state with mprotect while a waiter runs,
// and baits the registry with an entry planted while the waiter
// is parked mid-join. It pins the rule against future edits —
// the TODO above moves where reaping happens — rather than
// closing an open hole.
//
// Reaping here at all is what makes the retention bound hold for a
// module that bursts and then goes quiet: the spawn-path reaper
// only runs if another call ever comes.
struct FinishOnExit {
PlainLogosObject* self;
std::uint64_t id;
~FinishOnExit()
{
self->reapFinishedWaiters(); // others, never itself
self->publishFinishedWaiter(id); // strictly last
}
} finishOnExit{this, waiterId};
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));
return;
}
auto res = fut->get();
if (!res.ok) {
postToQtEventLoop(std::move(callback), QVariant(),
logos::callErrorFromWire(objectName, res.errCode,
res.err));
return;
}
QVariant value = rpcValueToQVariant(res.value);
// Resolve a "multi" provider's deferred completion (sentinel → wait for
// 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;
if (logos::isPendingCallSentinel(value, &callId))
value = awaitCompletion(callId, timeoutMs, methodName, &err);
}
postToQtEventLoop(std::move(callback), std::move(value), std::move(err));
});
m_waiters.emplace(waiterId, std::move(waiter));
}
}
bool PlainLogosObject::informModuleToken(const QString& authToken,
@@ -266,7 +699,16 @@ void PlainLogosObject::release()
// 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.
//
// 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 — 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. Waiters that already finished were reaped as the
// calls after them were issued; this collects whatever is left.
disconnectEvents();
stopAndJoinWaiters();
m_conn.reset();
delete this;
}
cpp/implementations/plain/plain_logos_object.h
+92 -2
View File
@@ -5,11 +5,14 @@
#include "rpc_connection.h"
#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <utility>
#include <vector>
@@ -23,7 +26,7 @@ namespace logos::plain {
// Owns a shared_ptr<RpcConnectionBase>; the transport layer hands the
// connection over after opening the socket. release() stops the connection.
// -----------------------------------------------------------------------------
class PlainLogosObject : public LogosObject {
class PlainLogosObject : public LogosObject, public LogosObjectErrorChannel {
public:
PlainLogosObject(std::string objectName,
std::shared_ptr<RpcConnectionBase> conn);
@@ -40,6 +43,21 @@ public:
int timeoutMs,
AsyncResultCallback callback) override;
// LogosObjectErrorChannel — the real implementations. The two LogosObject
// entry points above are thin adapters that discard the error, so there is
// exactly ONE call path per direction and the two front doors cannot drift.
QVariant callMethodWithError(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int timeoutMs,
logos::CallError* err) override;
void callMethodAsyncWithError(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int timeoutMs,
AsyncResultErrorCallback callback) override;
bool informModuleToken(const QString& authToken,
const QString& moduleName,
const QString& token,
@@ -61,7 +79,49 @@ 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();
QVariant awaitCompletion(const QString& callId, int timeoutMs);
// `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);
// 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();
// Join and drop the waiters that have already FINISHED, so a handle that
// outlives its calls does not accumulate them. TWO call sites, which
// between them cover both shapes of traffic:
//
// * every async spawn — a call pays for the corpses of earlier ones;
// * every waiter as it finishes, BEFORE it publishes its own id — so a
// burst drains itself instead of parking until the next call, which for
// a module that bursts and goes quiet may never come.
//
// NOT called from stopAndJoinWaiters(): teardown joins by id-independent
// brute force and needs no published list. (It used to say otherwise here;
// it never did.) Cheap either way: a join on an already-returned thread is
// a couple of syscalls, and only ids a waiter itself published are touched.
void reapFinishedWaiters();
// A waiter's FINAL act — see the scope guard in callMethodAsyncWithError.
// After this returns, that thread never touches the object again, which is
// what makes it safe for someone else to join and drop it. Nothing the
// waiter does may follow it, its own reap least of all.
void publishFinishedWaiter(std::uint64_t id);
std::string m_objectName;
std::shared_ptr<RpcConnectionBase> m_conn;
@@ -72,6 +132,36 @@ private:
std::condition_variable m_completionCv;
std::map<QString, QVariant> m_completions;
bool m_completionSubscribed = false;
// The waiter registry. KEYED, not a plain vector, because a thread cannot
// join itself: a waiter can therefore never retire its own entry, and a
// vector left only one moment to clear it — teardown — so every completed
// call parked a finished-but-unjoined thread (~one page of resident memory
// each) for the whole life of the handle. The production shape is one
// cached handle per module reused for every call (logos_api_consumer.cpp),
// so that grew without bound. Now a waiter publishes its id into
// m_finishedWaiters as its last act, and both the next spawn and every
// OTHER waiter on its way out join and erase it: see reapFinishedWaiters().
//
// Retention tracks neither call count nor peak concurrency. A burst drains
// as it completes, because each waiter reaps the ones that finished before
// it. What survives an idle handle is only what published after the last
// reap — at minimum the last waiter to finish, which by construction has
// nobody behind it to collect it (measured: 1-2 after a 2000-call burst).
// The next call, or teardown, takes those.
//
// The real fix is still the TODO in callMethodAsyncWithError — fold the
// wait into the shared Asio io_context and have no thread per pending RPC
// at all. This makes the interim honest, it does not replace that.
std::mutex m_waiterMu;
std::map<std::uint64_t, std::thread> m_waiters;
std::vector<std::uint64_t> m_finishedWaiters;
std::uint64_t m_nextWaiterId = 0;
// 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<bool> m_stopping{false};
};
} // namespace logos::plain
cpp/implementations/plain/plain_transport_host.cpp
+40
View File
@@ -11,7 +11,10 @@
#include <QMetaObject>
#include <atomic>
#include <chrono>
#include <future>
#include <boost/asio/post.hpp>
#include <boost/asio/ssl/context.hpp>
#include <boost/version.hpp>
#include <openssl/ssl.h>
@@ -210,6 +213,43 @@ PlainTransportHost::~PlainTransportHost()
}
if (tcp) tcp->stop();
if (ssl) ssl->stop();
// Quiesce the I/O thread before this host (an IncomingCallHandler) is
// destroyed. Server-side RpcConnections hold a RAW `IncomingCallHandler*`
// back to us; a read completion racing this teardown runs
// RpcConnection::fail() on the I/O thread, which calls
// m_handler->onConnectionClosed(this). stop() above closes the sockets but
// does NOT wait for an already-executing fail() — so without this barrier
// the handler can be freed mid-call (a use-after-free that surfaced as a
// flaky SIGSEGV, including on macOS CI in CallErrorAfterAcquireTest right
// after a preceding live-host test tore its PlainTransportHost down).
// There is a single shared I/O thread, so a task posted now runs only after
// every in-flight/queued connection handler has completed; blocking on it
// guarantees no callback still references this host. Skip when we're ON
// the I/O thread (the in-flight handler is our own caller) to avoid
// self-deadlock.
if (tcp || ssl) {
auto& ioc = IoContextPool::shared().ioContext();
if (!ioc.get_executor().running_in_this_thread()) {
// The promise is shared, NOT captured by reference. The wait below is
// bounded, so on timeout this frame returns while the posted task is
// still queued — a by-reference capture would then set_value() on a
// destroyed stack object, which is the very failure mode this barrier
// exists to prevent.
auto drained = std::make_shared<std::promise<void>>();
auto fut = drained->get_future();
boost::asio::post(ioc, [drained] { drained->set_value(); });
// Bounded so a wedged I/O thread cannot hang teardown — but a timeout
// means the barrier did NOT hold and we are about to free an
// IncomingCallHandler a connection may still call back into. Say so:
// silently proceeding is how this class of crash stays unexplained.
if (fut.wait_for(std::chrono::seconds(5)) != std::future_status::ready) {
qWarning() << "PlainTransportHost: I/O drain timed out after 5s;"
<< "tearing down anyway — a connection callback may still"
<< "reference this host (see the barrier comment above)";
}
}
}
}
bool PlainTransportHost::start()
cpp/implementations/qt_local/local_transport.cpp
+52 -11
View File
@@ -39,10 +39,12 @@ private:
} // anonymous namespace
class LocalLogosObject : public LogosObject {
class LocalLogosObject : public LogosObject, public LogosObjectErrorChannel {
public:
explicit LocalLogosObject(ModuleProxy* proxy)
: m_proxy(proxy), m_helper(nullptr)
// objectName is carried purely so a failure can name the module it belongs
// to (logos::CallError::origin).
LocalLogosObject(ModuleProxy* proxy, QString objectName)
: m_proxy(proxy), m_helper(nullptr), m_objectName(std::move(objectName))
{
qDebug() << "[LogosObject] Created LocalLogosObject wrapping ModuleProxy" << reinterpret_cast<quintptr>(proxy);
}
@@ -52,31 +54,62 @@ public:
delete m_helper;
}
// Adapters over the error-carrying implementations: they discard the
// diagnosis, which is exactly what these entry points have always done.
QVariant callMethod(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int /*timeoutMs*/) override
int timeoutMs) override
{
if (!m_proxy) return QVariant();
qDebug() << "[LogosObject] LocalLogosObject::callMethod" << methodName << "args:" << args.size();
return m_proxy->callRemoteMethod(authToken, methodName, args);
return callMethodWithError(authToken, methodName, args, timeoutMs, nullptr);
}
void callMethodAsync(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int /*timeoutMs*/,
int timeoutMs,
AsyncResultCallback callback) override
{
if (!callback) return;
callMethodAsyncWithError(authToken, methodName, args, timeoutMs,
[cb = std::move(callback)](QVariant v, const logos::CallError&) mutable {
cb(std::move(v));
});
}
// In-process direct dispatch: there is no wire to drop and no deadline to
// miss, so a vanished ModuleProxy is the only failure this transport has.
QVariant callMethodWithError(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int /*timeoutMs*/,
logos::CallError* err) override
{
if (err) err->clear();
if (!m_proxy) {
QTimer::singleShot(0, [callback]() { callback(QVariant()); });
if (err) *err = proxyGoneError();
return QVariant();
}
qDebug() << "[LogosObject] LocalLogosObject::callMethod" << methodName << "args:" << args.size();
return m_proxy->callRemoteMethod(authToken, methodName, args);
}
void callMethodAsyncWithError(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int /*timeoutMs*/,
AsyncResultErrorCallback callback) override
{
if (!callback) return;
if (!m_proxy) {
const logos::CallError e = proxyGoneError();
QTimer::singleShot(0, [callback, e]() { callback(QVariant(), e); });
return;
}
ModuleProxy* proxy = m_proxy;
QTimer::singleShot(0, [proxy, authToken, methodName, args, callback]() {
QVariant result = proxy->callRemoteMethod(authToken, methodName, args);
callback(result);
callback(result, logos::CallError{});
});
}
@@ -134,8 +167,16 @@ public:
quintptr id() const override { return reinterpret_cast<quintptr>(m_proxy); }
private:
logos::CallError proxyGoneError() const
{
const std::string origin = m_objectName.toStdString();
return logos::callErrorObjectUnavailable(
origin, "module '" + origin + "' is no longer registered locally");
}
ModuleProxy* m_proxy;
EventHelper* m_helper;
QString m_objectName;
};
// ── LocalTransportHost ───────────────────────────────────────────────────────
@@ -188,7 +229,7 @@ LogosObject* LocalTransportConnection::requestObject(const QString& objectName,
}
qDebug() << "[LogosObject] LocalTransportConnection: returning LocalLogosObject for:" << objectName;
return new LocalLogosObject(proxy);
return new LocalLogosObject(proxy, objectName);
}
#include "local_transport.moc"
cpp/implementations/qt_remote/remote_transport.cpp
+129 -23
View File
@@ -57,10 +57,13 @@ private:
} // anonymous namespace
class RemoteLogosObject : public LogosObject {
class RemoteLogosObject : public LogosObject, public LogosObjectErrorChannel {
public:
explicit RemoteLogosObject(QObject* replica)
: m_replica(replica), m_helper(nullptr)
// objectName is carried purely so a failure can name the module it belongs
// to — logos::CallError::origin, the same field the acquire-time error and
// lp_invoke's out_error_json already fill in.
RemoteLogosObject(QObject* replica, QString objectName)
: m_replica(replica), m_helper(nullptr), m_objectName(std::move(objectName))
{
qDebug() << "[LogosObject] Created RemoteLogosObject wrapping QRemoteObjectReplica" << reinterpret_cast<quintptr>(replica);
if (m_replica) {
@@ -96,7 +99,9 @@ public:
// first (mirrors the QPointer guard in invokeRemoteMethodAsync).
if (cb)
QTimer::singleShot(0, m_helper,
[cb = std::move(cb), result]() { cb(result); });
[cb = std::move(cb), result]() {
cb(result, logos::CallError{});
});
}
});
}
@@ -115,13 +120,32 @@ public:
}
}
// Adapter over the error-carrying implementation: discards the diagnosis,
// which is exactly what this entry point has always done.
QVariant callMethod(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int timeoutMs) override
{
return callMethodWithError(authToken, methodName, args, timeoutMs, nullptr);
}
QVariant callMethodWithError(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int timeoutMs,
logos::CallError* err) override
{
if (err) err->clear();
const std::string origin = m_objectName.toStdString();
const std::string method = methodName.toStdString();
if (!m_replica) {
qWarning() << "RemoteLogosObject: Cannot call method on null replica";
if (err)
*err = logos::callErrorObjectUnavailable(
origin, "replica for '" + origin + "' is gone (module unloaded"
" or transport dropped)");
return QVariant();
}
qDebug() << "[LogosObject] RemoteLogosObject::callMethod" << methodName << "args:" << args.size();
@@ -139,22 +163,41 @@ public:
if (!success) {
qWarning() << "RemoteLogosObject: Failed to invoke callRemoteMethod on replica";
if (err)
*err = logos::callErrorCallFailed(
origin, "replica did not accept callRemoteMethod for '"
+ method + "'");
return QVariant();
}
pendingCall.waitForFinished(timeoutMs);
if (!pendingCall.isFinished() || pendingCall.error() != QRemoteObjectPendingCall::NoError) {
qWarning() << "RemoteLogosObject: callRemoteMethod failed or timed out:" << pendingCall.error();
// Two distinct outcomes that used to collapse into one empty QVariant:
// the deadline elapsed with the call still in flight (timeout), and QtRO
// itself failed the call (transport).
if (!pendingCall.isFinished()) {
qWarning() << "RemoteLogosObject: callRemoteMethod timed out";
if (err) *err = logos::callErrorTimeout(origin, method, timeoutMs);
return QVariant();
}
if (pendingCall.error() != QRemoteObjectPendingCall::NoError) {
qWarning() << "RemoteLogosObject: callRemoteMethod failed:" << pendingCall.error();
if (err)
*err = logos::callErrorTransport(
origin, "QtRO call to '" + origin + "." + method
+ "' failed with error "
+ std::to_string(static_cast<int>(pendingCall.error())));
return QVariant();
}
// A "multi" provider may have deferred the result (returned a pending
// sentinel); resolveDeferred waits for the completion event, or returns
// the value unchanged for an ordinary (synchronous) result.
return resolveDeferred(pendingCall.returnValue(), timeoutMs);
return resolveDeferred(pendingCall.returnValue(), timeoutMs, methodName, err);
}
// Adapter over the error-carrying implementation: discards the diagnosis,
// which is exactly what this entry point has always done.
void callMethodAsync(const QString& authToken,
const QString& methodName,
const QVariantList& args,
@@ -162,8 +205,27 @@ public:
AsyncResultCallback callback) override
{
if (!callback) return;
callMethodAsyncWithError(authToken, methodName, args, timeoutMs,
[cb = std::move(callback)](QVariant v, const logos::CallError&) mutable {
cb(std::move(v));
});
}
void callMethodAsyncWithError(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int timeoutMs,
AsyncResultErrorCallback callback) override
{
if (!callback) return;
const std::string origin = m_objectName.toStdString();
const std::string method = methodName.toStdString();
if (!m_replica) {
QTimer::singleShot(0, [callback]() { callback(QVariant()); });
const logos::CallError e = logos::callErrorObjectUnavailable(
origin, "replica for '" + origin + "' is gone (module unloaded"
" or transport dropped)");
QTimer::singleShot(0, [callback, e]() { callback(QVariant(), e); });
return;
}
@@ -182,7 +244,9 @@ public:
if (!success) {
qWarning() << "RemoteLogosObject: Failed to invoke callRemoteMethod on replica (async)";
QTimer::singleShot(0, [callback]() { callback(QVariant()); });
const logos::CallError e = logos::callErrorCallFailed(
origin, "replica did not accept callRemoteMethod for '" + method + "'");
QTimer::singleShot(0, [callback, e]() { callback(QVariant(), e); });
return;
}
@@ -193,41 +257,76 @@ public:
auto* timer = new QTimer(watcher);
timer->setSingleShot(true);
// Exactly-once gate. The finished handler and the timeout timer can
// both be queued around the same moment; without a guard that races
// into a double callback, which violates callMethodAsyncWithError's
// contract and is a latent double-free for every consumer. The same
// gate wraps the deferred-completion path so a late completion cannot
// deliver after the initial timeout already has (or vice versa).
auto delivered = std::make_shared<std::atomic_bool>(false);
AsyncResultErrorCallback deliverOnce =
[delivered, callback = std::move(callback)](QVariant result,
const logos::CallError& err) mutable {
if (delivered->exchange(true))
return;
if (callback)
callback(std::move(result), err);
};
// Success handler -- delivers result on the consumer's thread
QObject::connect(watcher, &QRemoteObjectPendingCallWatcher::finished,
watcher, [this, callback, timer, timeoutMs](QRemoteObjectPendingCallWatcher* w) {
watcher, [this, deliverOnce, timer, timeoutMs, origin, method, delivered](QRemoteObjectPendingCallWatcher* w) {
timer->stop(); // cancel timeout
// Timeout may already have won the race and deleteLater'd us; if
// the slot still runs, do not enter the deferred-completion path
// or we would arm a second delivery after the caller already saw
// a timeout.
if (delivered->load()) {
w->deleteLater();
return;
}
QVariant result;
logos::CallError err;
if (w->error() == QRemoteObjectPendingCall::NoError) {
result = w->returnValue();
} else {
qWarning() << "RemoteLogosObject: async callMethod error:" << w->error();
err = logos::callErrorTransport(
origin, "QtRO call to '" + origin + "." + method
+ "' failed with error "
+ std::to_string(static_cast<int>(w->error())));
}
w->deleteLater();
// A "multi" provider may have deferred the result: wait for the
// completion event instead of delivering the pending sentinel.
{
if (err.ok()) {
QString callId;
if (logos::isPendingCallSentinel(result, &callId)) {
if (m_completions.contains(callId)) { callback(m_completions.take(callId)); return; }
m_asyncCompletionCbs.insert(callId, callback);
// Bound the wait: deliver an empty result once if it never lands.
QTimer::singleShot(timeoutMs, m_helper, [this, callId]() {
if (m_completions.contains(callId)) {
deliverOnce(m_completions.take(callId), logos::CallError{});
return;
}
m_asyncCompletionCbs.insert(callId, deliverOnce);
// Bound the wait: a completion that never lands is a timeout,
// reported as one instead of as an empty result.
QTimer::singleShot(timeoutMs, m_helper, [this, callId, origin, method, timeoutMs]() {
if (m_asyncCompletionCbs.contains(callId)) {
auto cb = m_asyncCompletionCbs.take(callId);
if (cb) cb(QVariant());
if (cb)
cb(QVariant(),
logos::callErrorTimeout(origin, method, timeoutMs));
}
});
return;
}
}
callback(result);
deliverOnce(result, err);
}, Qt::QueuedConnection);
// Timeout handler -- stops the watcher and delivers empty result
QObject::connect(timer, &QTimer::timeout, watcher, [watcher, callback]() {
// Timeout handler -- stops the watcher and reports the elapsed deadline
QObject::connect(timer, &QTimer::timeout, watcher, [watcher, deliverOnce, origin, method, timeoutMs]() {
qWarning() << "RemoteLogosObject: async callMethod timed out";
callback(QVariant());
deliverOnce(QVariant(), logos::callErrorTimeout(origin, method, timeoutMs));
watcher->deleteLater(); // also destroys the timer (child)
});
@@ -357,7 +456,9 @@ private:
// Resolve a possibly-deferred result. If `rv` is a pending sentinel from a
// "multi" provider, wait (up to timeoutMs) for the completion event keyed by
// callId, pumping the consumer event loop; otherwise return `rv` unchanged.
QVariant resolveDeferred(const QVariant& rv, int timeoutMs)
QVariant resolveDeferred(const QVariant& rv, int timeoutMs,
const QString& methodName = QString(),
logos::CallError* err = nullptr)
{
QString callId;
if (!logos::isPendingCallSentinel(rv, &callId)) return rv;
@@ -373,6 +474,10 @@ private:
m_completionWaiters.remove(callId);
if (m_completions.contains(callId)) return m_completions.take(callId);
qWarning() << "RemoteLogosObject: deferred call" << callId << "timed out";
if (err)
*err = logos::callErrorTimeout(m_objectName.toStdString(),
methodName.toStdString(),
timeoutMs > 0 ? timeoutMs : 30000);
return QVariant();
}
@@ -383,7 +488,8 @@ private:
// Touched only on the consumer event-loop thread.
QHash<QString, QVariant> m_completions;
QHash<QString, QEventLoop*> m_completionWaiters;
QHash<QString, AsyncResultCallback> m_asyncCompletionCbs;
QHash<QString, AsyncResultErrorCallback> m_asyncCompletionCbs;
QString m_objectName;
};
// ── RemoteTransportHost ──────────────────────────────────────────────────────
@@ -538,7 +644,7 @@ LogosObject* RemoteTransportConnection::requestObject(const QString& objectName,
qDebug() << "[LogosObject] RemoteTransportConnection: returning RemoteLogosObject for:" << objectName;
g_acquireCount.fetch_add(1, std::memory_order_relaxed);
return new RemoteLogosObject(replica);
return new RemoteLogosObject(replica, objectName);
}
long RemoteTransportConnection::acquireCount() { return g_acquireCount.load(std::memory_order_relaxed); }
cpp/logos_api_consumer.cpp
+27
View File
@@ -141,6 +141,16 @@ QVariant LogosAPIConsumer::invokeRemoteMethod(const QString& authToken, const QS
qDebug() << "[LogosObject] LogosAPIConsumer: calling via LogosObject::callMethod" << methodName;
// No release() here: the handle stays cached for the next call. Released in
// clearObjectCache() (destructor / reconnect) or evicted when stale.
//
// Prefer the error channel when the transport implements it (see
// LogosObjectErrorChannel in logos_object.h). Without it, `err` could only
// ever describe an ACQUIRE failure — everything that went wrong after the
// handle existed (the deadline elapsing, the connection dropping, the peer
// answering "not published") came back as a bare QVariant() with a clean
// err, i.e. reported as a method that returned null.
if (auto* channel = dynamic_cast<LogosObjectErrorChannel*>(plugin))
return channel->callMethodWithError(authToken, methodName, args,
timeout.ms, err);
return plugin->callMethod(authToken, methodName, args, timeout.ms);
}
@@ -216,6 +226,23 @@ void LogosAPIConsumer::invokeRemoteMethodAsync(const QString& authToken, const Q
// before the transport callback fires, the callback is silently dropped and
// the handle is released by the destructor's clearObjectCache(), not here.
QPointer<LogosAPIConsumer> self(this);
// Prefer the error channel when the transport implements it. The lambda
// below used to take only `QVariant result` and hand the caller a
// hard-coded empty logos::CallError — so once acquire had succeeded, every
// async outcome was reported as a success, whatever actually happened.
if (auto* channel = dynamic_cast<LogosObjectErrorChannel*>(plugin)) {
channel->callMethodAsyncWithError(authToken, methodName, args, timeout.ms,
[callback, self](QVariant result, const logos::CallError& err) {
if (!self)
return;
callback(std::move(result), err);
});
return;
}
// Transport without an error channel (the mock): unchanged behaviour —
// the value, and no diagnosis to give.
plugin->callMethodAsync(authToken, methodName, args, timeout.ms,
[callback, self](QVariant result) {
if (!self)
cpp/logos_call_error.h
+69 -2
View File
@@ -16,8 +16,22 @@ namespace logos {
// provider dispatch errors) without an ABI break.
//
// Currently produced:
// "object_unavailable" — the target module/object could not be acquired
// (not loaded, not published, or transport failure).
// "object_unavailable" — the target module/object is not there: it could not
// be acquired, or the transport answered that it is
// not published. One code for one condition, whether
// it is detected at acquire time (QtRO, which resolves
// a replica up front) or at call time (the plain wire,
// whose requestObject hands back a handle for any name
// over an open connection and only learns the truth
// from the reply).
// "timeout" — the caller's deadline elapsed with no reply.
// "transport_error" — the connection failed or was torn down mid-call.
// "call_failed" — the peer could not dispatch the call at all (as
// distinct from a provider that ran and REJECTED it,
// which answers the "dispatch_failed" envelope as its
// result value — see logos-cpp-sdk#129).
// "unauthorized" — the provider rejected our token and the one
// permitted re-exchange also failed.
struct CallError {
std::string code; // empty = no error
std::string message;
@@ -27,6 +41,59 @@ struct CallError {
void clear() { code.clear(); message.clear(); origin.clear(); }
};
// ---------------------------------------------------------------------------
// Canonical constructors.
//
// Every transport that can detect one of these produces it HERE rather than
// spelling the code out at the failure site, so a caller decoding a
// {code, message, origin} object gets the same vocabulary no matter which wire
// the call went over — and so lp_invoke and lp_invoke_async, which both render
// this struct with the same makeErrorJson(), stay indistinguishable.
// ---------------------------------------------------------------------------
inline CallError callErrorTimeout(const std::string& origin,
const std::string& method, int timeoutMs)
{
return {"timeout",
"call to '" + origin + "." + method + "' timed out after "
+ std::to_string(timeoutMs) + "ms",
origin};
}
inline CallError callErrorObjectUnavailable(const std::string& origin,
const std::string& detail)
{
return {"object_unavailable", detail, origin};
}
inline CallError callErrorTransport(const std::string& origin,
const std::string& detail)
{
return {"transport_error", detail, origin};
}
inline CallError callErrorCallFailed(const std::string& origin,
const std::string& detail)
{
return {"call_failed", detail, origin};
}
// Map a plain-wire ResultMessage failure (errCode + err) onto the vocabulary
// above. The wire's codes are the transport's own spelling; this is the single
// place they are translated, so a new wire code degrades to "call_failed"
// instead of silently becoming an empty CallError (which reads as SUCCESS).
inline CallError callErrorFromWire(const std::string& origin,
const std::string& wireCode,
const std::string& message)
{
const std::string detail = message.empty() ? wireCode : message;
if (wireCode == "MODULE_NOT_LOADED")
return callErrorObjectUnavailable(origin, detail);
if (wireCode == "TRANSPORT_CLOSED" || wireCode == "TRANSPORT_ERROR")
return callErrorTransport(origin, detail);
return callErrorCallFailed(origin, detail);
}
} // namespace logos
#endif // LOGOS_CALL_ERROR_H
cpp/logos_object.h
+69
View File
@@ -1,6 +1,8 @@
#ifndef LOGOS_OBJECT_H
#define LOGOS_OBJECT_H
#include "logos_call_error.h"
#include <QString>
#include <QVariant>
#include <QVariantList>
@@ -126,4 +128,71 @@ public:
virtual bool isValid() const { return true; }
};
/**
* @brief Optional extension: calls that report WHY they failed.
*
* LogosObject's own callMethod/callMethodAsync answer a bare QVariant() for
* every failure — a timeout, a torn-down connection, and a module that is not
* published all look identical to a provider that legitimately returned null.
* That is the whole reason lp_invoke and lp_invoke_async could report success
* for a call that never happened.
*
* This interface is DELIBERATELY a sibling of LogosObject rather than more
* virtuals on it. LogosObject is an installed header (`include/logos_object.h`)
* whose vtable is baked into every statically-linked copy of liblogos_protocol
* in a process — one per loaded module, each pinned to its own protocol
* revision. Appending a virtual would append a vtable slot, and a caller
* compiled against the new header calling that slot on an object whose vtable
* came from an older copy is undefined behaviour. Declaring a separate
* interface and reaching it with dynamic_cast leaves LogosObject's layout,
* size and vtable byte-for-byte unchanged, so no such pairing can exist:
* a copy that does not know about this interface simply fails the cast.
*
* Consumers therefore MUST treat it as optional:
*
* if (auto* ch = dynamic_cast<LogosObjectErrorChannel*>(obj))
* ch->callMethodWithError(...); // real diagnosis
* else
* obj->callMethod(...); // today's behaviour, unchanged
*
* Implemented by the plain (tcp/tcp_ssl), qt_remote (QtRO) and qt_local
* transports. NOT implemented by the mock transport: MockStore always answers,
* so there is no failure to report, and leaving MockLogosObject alone keeps the
* one subclass whose header is installed (implementations/mock/mock_transport.h)
* layout-identical too.
*/
class LogosObjectErrorChannel {
public:
virtual ~LogosObjectErrorChannel() = default;
/**
* @brief callMethod, plus the reason on failure.
* @param err Cleared on entry; set to the canonical {code, message, origin}
* on failure. May be null (then this is exactly callMethod).
* @return The method result, or an invalid QVariant on failure.
*/
virtual QVariant callMethodWithError(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int timeoutMs,
logos::CallError* err) = 0;
using AsyncResultErrorCallback =
std::function<void(QVariant, const logos::CallError&)>;
/**
* @brief callMethodAsync, whose callback carries the reason on failure.
*
* Same delivery contract as LogosObject::callMethodAsync: the callback
* fires on a subsequent event-loop iteration, never synchronously, and
* exactly once. On success the error argument is a default-constructed
* (ok()) CallError.
*/
virtual void callMethodAsyncWithError(const QString& authToken,
const QString& methodName,
const QVariantList& args,
int timeoutMs,
AsyncResultErrorCallback callback) = 0;
};
#endif // LOGOS_OBJECT_H
cpp/logos_protocol.h
+16 -3
View File
@@ -193,9 +193,22 @@ int lp_invoke(lp_client* client,
* `cb` carries the same outcome the sync twin splits across its return code
* and out-params: ok != 0 → `json` is the result JSON value; ok == 0 → `json`
* is the canonical error object lp_invoke would have written to
* out_error_json (e.g. code "object_unavailable" when the target module is
* not loaded). A LP_OK return therefore means "dispatched", never "succeeded"
* — the outcome is only known in the callback.
* out_error_json. A LP_OK return therefore means "dispatched", never
* "succeeded" — the outcome is only known in the callback.
*
* WHAT ok == 0 COVERS, precisely, because "the same outcome as the sync twin"
* is a statement about PARITY and not about completeness. Reported: failure to
* acquire the target ("object_unavailable"), a call that exceeds its deadline,
* a rejected auth token, and MODULE_NOT_LOADED from a host that is up. Both
* twins report all four; neither did before.
*
* NOT reported, and it is not an oversight: an unknown method name. Every
* provider flavour answers one with a bare null, byte-identical to a method
* that legitimately returns null, so the distinction does not exist on the
* wire to be reported. Closing it needs a provider-contract change across the
* SDKs, not a transport change here. A provider's own rejection of well-formed
* arguments ("dispatch_failed") is likewise NOT folded in by either twin — it
* arrives as a result, and the generated wrappers fold it.
*
* Argument/handle validation still fails synchronously with
* LP_ERR_INVALID_ARG and `cb` is NOT called in that case.