mirror of
https://github.com/logos-co/logos-protocol.git
synced 2026-08-30 21:41:10 +00:00
Every LiveHost fixture in tests/protocol (five of them: test_iofold,
test_plain_object_teardown, test_plain_waiter_reaping,
test_plain_completion_sub_lifetime, test_call_error_after_acquire) puts a
ModuleProxy on a worker QThread and publishes it through a PlainTransportHost,
and every one of them freed that host from the TEST thread:
m_host.reset(); // test thread
m_thread->quit();
m_thread->wait();
That is a use-after-free with a millisecond-wide window, in code five test
files share.
WHY. publishObject() connects a lambda to the proxy's eventResponse signal with
NO context object, so it is a direct connection and runs on whichever thread
emits. ModuleProxy always QUEUES that emission to its own thread — it must, or
QtRO source serialization races the reply socket — so the emitting thread is
always the worker. The lambda converts the payload and then calls fanOutEvent,
which locks the host's m_mu. Free the host on the test thread and that lock is
on a destroyed mutex. ~PlainTransportHost does disconnect the connection, which
covers an emission that has not started; a worker already inside the lambda is
not called back, it is simply running, and qvariantListToRpcList on a real
payload sits in front of the lock.
MOVING reset() AFTER quit()/wait() IS NOT THE FIX, and the obvious reason for
saying so is wrong, so here is the measured one. That order does stop the fault:
wait() joins the worker, so an in-flight lambda has finished, and it is clean
under Guard Malloc. It is clean because it throws the queue away —
QThread::quit() reaches QEventLoop::exit(), which sets the exit flag
SYNCHRONOUSLY from the calling thread instead of posting an event, so the
worker's loop stops at its next iteration and discards every emission still
queued behind it. Same specimen, same load: 273-383 of 960 events delivered,
silently. In a suite whose tests count deliveries that is the worse failure,
because nothing reports it. It also shuts the host down AFTER the proxy's
thread, which test_call_error_after_acquire needs the other way round.
THE FIX, in one shared place (live_host_teardown.h) rather than five copies,
because a copied pattern is what this was: destroy the host ON the proxy's
thread, via the SDK's own logos::runOnOwnerThread marshal. A QMetaCallEvent is
dispatched by the worker's event loop, so while it runs the worker is by
definition not inside any other slot. Qt dispatches equal-priority events FIFO,
so every emission queued before it runs first, against a live host; everything
after finds the connection already severed by ~PlainTransportHost. The io
thread, the third thread that reaches the host, is still covered by the drain
barrier ~PlainTransportHost already carries — the proxy thread is not the io
thread, so that barrier's running_in_this_thread() check still takes the
blocking path. The added blocking wait introduces no hang that was not already
there: the next two statements are quit()/wait() on the same thread, with no
timeout.
EVIDENCE. test_plain_host_event_teardown.cpp is the specimen, and it is a
detector: 24 wide events queued per round, teardown aimed at the trailing edge
of the first so the worker is inside the host's lambda. Validated the way this
directory validates detectors — against a real checkout of the code it replaces
(cf1b9b0), with the fixture's teardown as it was:
pre-fix, no detector: SIGABRT 3/3 runs ("mutex lock failed: Invalid
argument" out of a Qt event handler)
pre-fix, Guard Malloc: SIGSEGV 3/3 runs at plain_transport_host.cpp:354
in fanOutEvent, on the thread named "QThread", under
ModuleProxy::eventResponse
naive (quit/wait/reset): NO fault, either detector — and 273-383 of 960
emissions delivered; the rest silently dropped
fixed, no detector: 960/960 emitted, 0 after the free, 40/40 rounds
still draining when teardown began
fixed, Guard Malloc: same counts, clean
WHAT THIS IS NOT. The five fixtures do not fault on their own today: pre-fix,
their suites are clean under Guard Malloc (2/2 runs, 33 tests) because their
test bodies drain the burst before teardown. So this closes a live trap in
shared fixture code — one that two separate probes fell into by copying the
pattern — rather than a reproduced CI failure. It remains a candidate for the
suite's unexplained SIGSEGVs, not a proven cause, and it is stated that way in
the test file.
Full suite: 304/304 (302 before, +2 new). `nix build .#tests`: 304/304 via
ctest, 95s. The five affected suites are clean under Guard Malloc on the fixed
tree.
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
412 lines
15 KiB
C++
412 lines
15 KiB
C++
// The call-error channel AFTER the target has been acquired.
|
|
//
|
|
// logos-protocol#40 made lp_invoke_async able to report a failure at all, but
|
|
// only for the two conditions the layers ABOVE the transport produce: acquire
|
|
// failure ("object_unavailable") and the unauthorized sentinel. Everything the
|
|
// transport itself learns while the call is in flight was still discarded:
|
|
//
|
|
// * PlainLogosObject::callMethod / callMethodAsync answer a bare QVariant()
|
|
// for BOTH `future timed out` and `ResultMessage.ok == false` — throwing
|
|
// away res.err / res.errCode, which the wire already carries;
|
|
// * LogosAPIConsumer::invokeRemoteMethodAsync then hard-coded an empty
|
|
// logos::CallError next to that value.
|
|
//
|
|
// So once acquire succeeded, both entry points reported success no matter what
|
|
// happened. Two conditions in particular are ordinary, not exotic:
|
|
//
|
|
// * a TIMEOUT — the caller's own deadline elapsed and nothing came back;
|
|
// * MODULE NOT LOADED against a LIVE host. PlainTransportConnection::
|
|
// requestObject never checks publication (it just constructs a handle over
|
|
// the open connection), so "the module isn't there" is NOT an acquire
|
|
// failure on this transport — it is a MODULE_NOT_LOADED ResultMessage at
|
|
// call time, and #40's object_unavailable never fires for it.
|
|
//
|
|
// These tests are a matched set and only mean something together: the two
|
|
// failures must report ok=0 / LP_ERR_UNAVAILABLE with a canonical
|
|
// {code,message,origin} object, and the successful control must still report
|
|
// ok=1 with its value — a fix that reported failure everywhere would satisfy
|
|
// the first half and break the second.
|
|
//
|
|
// Sync and async are BOTH covered because both had the identical hole: an
|
|
// async-only fix would leave lp_invoke lying while lp_invoke_async told the
|
|
// truth, which is the opposite of the parity #40 set out to establish.
|
|
//
|
|
// Everything runs against a real transport (plain TCP) and a live in-process
|
|
// PlainTransportHost, never the mock.
|
|
|
|
#include <gtest/gtest.h>
|
|
|
|
#include "logos_protocol.h"
|
|
|
|
#include "logos_provider_interface.h"
|
|
#include "logos_transport_config.h"
|
|
#include "module_proxy.h"
|
|
|
|
#include "plain_transport_host.h"
|
|
|
|
#include "live_host_teardown.h"
|
|
|
|
#include <QCoreApplication>
|
|
#include <QElapsedTimer>
|
|
#include <QJsonArray>
|
|
#include <QString>
|
|
#include <QThread>
|
|
#include <QVariant>
|
|
#include <QVariantList>
|
|
|
|
#include <nlohmann/json.hpp>
|
|
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstdint>
|
|
#include <iostream>
|
|
#include <memory>
|
|
#include <string>
|
|
#include <thread>
|
|
|
|
using namespace logos::plain;
|
|
|
|
namespace {
|
|
|
|
// A provider faithful to the real ones: compute() answers 7, slow() blocks
|
|
// past any sane deadline, and an UNKNOWN method answers a bare QVariant() —
|
|
// which is exactly what logos-qt-sdk's QtProviderObject and every generated
|
|
// provider dispatch do for a name they don't recognise.
|
|
class SlowProvider : public LogosProviderObject {
|
|
public:
|
|
QVariant callMethod(const QString& method, const QVariantList& args) override
|
|
{
|
|
if (method == QLatin1String("compute")) return QVariant(7);
|
|
if (method == QLatin1String("echo")) return args.value(0);
|
|
if (method == QLatin1String("slow")) {
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(3000));
|
|
return QVariant(1);
|
|
}
|
|
return QVariant(); // unknown method — indistinguishable from null
|
|
}
|
|
QJsonArray getMethods() override { return QJsonArray{}; }
|
|
bool informModuleToken(const QString&, const QString&) override { return true; }
|
|
void setEventListener(EventCallback) override {}
|
|
void init(void*) override {}
|
|
QString providerName() const override { return QStringLiteral("slow"); }
|
|
QString providerVersion() const override { return QStringLiteral("1.0.0"); }
|
|
};
|
|
|
|
QCoreApplication* ensureApp()
|
|
{
|
|
static int argc = 0;
|
|
static char* argv[] = { nullptr };
|
|
if (!QCoreApplication::instance())
|
|
new QCoreApplication(argc, argv);
|
|
return QCoreApplication::instance();
|
|
}
|
|
|
|
struct Capture {
|
|
std::atomic<bool> fired{false};
|
|
int ok = -1;
|
|
std::string json;
|
|
};
|
|
|
|
void captureCb(int ok, const char* json, void* user_data)
|
|
{
|
|
auto* c = static_cast<Capture*>(user_data);
|
|
c->ok = ok;
|
|
c->json = json ? json : "";
|
|
c->fired = true;
|
|
}
|
|
|
|
bool pumpUntilFired(Capture& c, int budgetMs)
|
|
{
|
|
QElapsedTimer timer;
|
|
timer.start();
|
|
while (!c.fired && timer.elapsed() < budgetMs)
|
|
QCoreApplication::processEvents(QEventLoop::AllEvents, 20);
|
|
return c.fired;
|
|
}
|
|
|
|
// A live host publishing `slow_module` through a ModuleProxy that lives on its
|
|
// OWN thread. The worker thread matters: PlainTransportHost::onCall dispatches
|
|
// to the proxy's thread, so a provider that sleeps would otherwise block the
|
|
// very event loop the consumer needs to deliver its own callback, and the test
|
|
// would measure the harness rather than the protocol.
|
|
class LiveHost {
|
|
public:
|
|
LiveHost()
|
|
{
|
|
LogosTransportConfig cfg;
|
|
cfg.protocol = LogosProtocol::Tcp;
|
|
cfg.host = "127.0.0.1";
|
|
cfg.port = 0; // ephemeral
|
|
m_host = std::make_unique<PlainTransportHost>(cfg);
|
|
m_started = m_host->start();
|
|
|
|
m_proxy = new ModuleProxy(&m_provider);
|
|
m_proxy->saveToken(QStringLiteral("origin"), QStringLiteral("live-token"));
|
|
m_thread = new QThread;
|
|
m_proxy->moveToThread(m_thread);
|
|
m_thread->start();
|
|
|
|
m_published = m_host->publishObject("slow_module", m_proxy);
|
|
|
|
const QString endpoint = m_host->endpoint();
|
|
m_port = endpoint.mid(endpoint.lastIndexOf(':') + 1).toUShort();
|
|
}
|
|
|
|
~LiveHost()
|
|
{
|
|
// Order matters: tear the host down FIRST so no inbound frame can be
|
|
// dispatched to the proxy while we are dismantling it, then stop the
|
|
// proxy's thread, then delete the proxy it was serving. The teardown
|
|
// itself runs on the PROXY's thread — see live_host_teardown.h.
|
|
logos::testing::destroyHostOnProxyThread(m_host, m_proxy);
|
|
QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
|
|
m_thread->quit();
|
|
m_thread->wait();
|
|
delete m_proxy;
|
|
delete m_thread;
|
|
}
|
|
|
|
bool ok() const { return m_started && m_published && m_port != 0; }
|
|
|
|
std::string target() const
|
|
{
|
|
return "{\"protocol\":\"tcp\",\"host\":\"127.0.0.1\",\"port\":"
|
|
+ std::to_string(m_port) + "}";
|
|
}
|
|
|
|
private:
|
|
SlowProvider m_provider;
|
|
std::unique_ptr<PlainTransportHost> m_host;
|
|
ModuleProxy* m_proxy = nullptr;
|
|
QThread* m_thread = nullptr;
|
|
bool m_started = false;
|
|
bool m_published = false;
|
|
uint16_t m_port = 0;
|
|
};
|
|
|
|
// Create an lp_client for `module` at `endpoint`, with its token pre-saved so
|
|
// the capability_module handshake is skipped and only the call path is under
|
|
// test.
|
|
lp_client* clientFor(const char* module, const std::string& endpoint)
|
|
{
|
|
lp_token_save(module, "live-token");
|
|
return lp_client_create(module, "origin", endpoint.c_str(), endpoint.c_str());
|
|
}
|
|
|
|
void destroyClient(lp_client* c)
|
|
{
|
|
lp_client_destroy(c);
|
|
QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
|
|
}
|
|
|
|
} // namespace
|
|
|
|
class CallErrorAfterAcquireTest : public ::testing::Test {
|
|
protected:
|
|
void SetUp() override { ensureApp(); }
|
|
};
|
|
|
|
// ── control: a successful async call still reports its value ────────────────
|
|
TEST_F(CallErrorAfterAcquireTest, AsyncSuccessStillReportsTheValue)
|
|
{
|
|
LiveHost host;
|
|
ASSERT_TRUE(host.ok());
|
|
|
|
lp_client* client = clientFor("slow_module", host.target());
|
|
ASSERT_NE(client, nullptr);
|
|
|
|
Capture c;
|
|
ASSERT_EQ(lp_invoke_async(client, "compute", "[]", 5000, &captureCb, &c), LP_OK);
|
|
ASSERT_TRUE(pumpUntilFired(c, 15000)) << "async callback never fired";
|
|
|
|
std::cout << " ASYNC success -> ok=" << c.ok << " json=" << c.json << std::endl;
|
|
|
|
EXPECT_EQ(c.ok, 1) << "a successful async call reported failure";
|
|
nlohmann::json v = nlohmann::json::parse(c.json, nullptr, false);
|
|
ASSERT_TRUE(v.is_number());
|
|
EXPECT_DOUBLE_EQ(v.get<double>(), 7.0);
|
|
|
|
destroyClient(client);
|
|
}
|
|
|
|
// ── the caller's deadline elapsed ───────────────────────────────────────────
|
|
//
|
|
// The provider sleeps 3s; the call is given 600ms. Pre-fix this delivered
|
|
// ok=1 with json "null" — a timeout dressed up as a provider that returned
|
|
// nothing.
|
|
TEST_F(CallErrorAfterAcquireTest, AsyncTimeoutReportsTheError)
|
|
{
|
|
LiveHost host;
|
|
ASSERT_TRUE(host.ok());
|
|
|
|
lp_client* client = clientFor("slow_module", host.target());
|
|
ASSERT_NE(client, nullptr);
|
|
|
|
Capture c;
|
|
ASSERT_EQ(lp_invoke_async(client, "slow", "[]", 600, &captureCb, &c), LP_OK);
|
|
ASSERT_TRUE(pumpUntilFired(c, 15000)) << "async callback never fired";
|
|
|
|
std::cout << " ASYNC timeout -> ok=" << c.ok << " json=" << c.json << std::endl;
|
|
|
|
EXPECT_EQ(c.ok, 0) << "a timed-out async call reported success";
|
|
nlohmann::json e = nlohmann::json::parse(c.json, nullptr, false);
|
|
ASSERT_TRUE(e.is_object()) << "ok==0 must carry the canonical error object";
|
|
EXPECT_EQ(e.value("code", std::string{}), "timeout");
|
|
EXPECT_EQ(e.value("origin", std::string{}), "slow_module");
|
|
EXPECT_FALSE(e.value("message", std::string{}).empty());
|
|
|
|
destroyClient(client);
|
|
// The provider is still sleeping; let it drain before the host goes away.
|
|
QElapsedTimer t; t.start();
|
|
while (t.elapsed() < 3500) QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
|
|
}
|
|
|
|
// ── the module is not loaded, on a host that IS up ──────────────────────────
|
|
//
|
|
// The single most common real failure, and the one #40's release notes claim
|
|
// to have fixed. It does NOT go through the acquire path on this transport:
|
|
// PlainTransportConnection::requestObject hands back a handle for any name over
|
|
// an open connection, so the failure surfaces as a MODULE_NOT_LOADED
|
|
// ResultMessage at call time — which was discarded.
|
|
TEST_F(CallErrorAfterAcquireTest, AsyncModuleNotLoadedOnALiveHostReportsTheError)
|
|
{
|
|
LiveHost host;
|
|
ASSERT_TRUE(host.ok());
|
|
|
|
lp_client* client = clientFor("ghost_module", host.target());
|
|
ASSERT_NE(client, nullptr);
|
|
|
|
Capture c;
|
|
ASSERT_EQ(lp_invoke_async(client, "compute", "[]", 5000, &captureCb, &c), LP_OK);
|
|
ASSERT_TRUE(pumpUntilFired(c, 15000)) << "async callback never fired";
|
|
|
|
std::cout << " ASYNC not-published -> ok=" << c.ok << " json=" << c.json << std::endl;
|
|
|
|
EXPECT_EQ(c.ok, 0) << "a call to an unpublished module reported success";
|
|
nlohmann::json e = nlohmann::json::parse(c.json, nullptr, false);
|
|
ASSERT_TRUE(e.is_object());
|
|
EXPECT_EQ(e.value("code", std::string{}), "object_unavailable");
|
|
EXPECT_EQ(e.value("origin", std::string{}), "ghost_module");
|
|
|
|
destroyClient(client);
|
|
}
|
|
|
|
// ── the synchronous twin had the identical hole ─────────────────────────────
|
|
TEST_F(CallErrorAfterAcquireTest, SyncSuccessStillReportsTheValue)
|
|
{
|
|
LiveHost host;
|
|
ASSERT_TRUE(host.ok());
|
|
|
|
lp_client* client = clientFor("slow_module", host.target());
|
|
ASSERT_NE(client, nullptr);
|
|
|
|
char* result = nullptr;
|
|
char* error = nullptr;
|
|
const int rc = lp_invoke(client, "compute", "[]", 5000, &result, &error);
|
|
|
|
std::cout << " SYNC success -> rc=" << rc
|
|
<< " result=" << (result ? result : "(null)")
|
|
<< " error=" << (error ? error : "(null)") << std::endl;
|
|
|
|
EXPECT_EQ(rc, LP_OK);
|
|
ASSERT_NE(result, nullptr);
|
|
EXPECT_STREQ(result, "7");
|
|
EXPECT_EQ(error, nullptr);
|
|
|
|
lp_string_free(result);
|
|
lp_string_free(error);
|
|
destroyClient(client);
|
|
}
|
|
|
|
TEST_F(CallErrorAfterAcquireTest, SyncTimeoutReportsTheError)
|
|
{
|
|
LiveHost host;
|
|
ASSERT_TRUE(host.ok());
|
|
|
|
lp_client* client = clientFor("slow_module", host.target());
|
|
ASSERT_NE(client, nullptr);
|
|
|
|
char* result = nullptr;
|
|
char* error = nullptr;
|
|
const int rc = lp_invoke(client, "slow", "[]", 600, &result, &error);
|
|
|
|
std::cout << " SYNC timeout -> rc=" << rc
|
|
<< " result=" << (result ? result : "(null)")
|
|
<< " error=" << (error ? error : "(null)") << std::endl;
|
|
|
|
EXPECT_EQ(rc, LP_ERR_UNAVAILABLE) << "a timed-out sync call reported success";
|
|
ASSERT_NE(error, nullptr) << "LP_ERR_UNAVAILABLE must carry the error object";
|
|
nlohmann::json e = nlohmann::json::parse(error, nullptr, false);
|
|
ASSERT_TRUE(e.is_object());
|
|
EXPECT_EQ(e.value("code", std::string{}), "timeout");
|
|
|
|
lp_string_free(result);
|
|
lp_string_free(error);
|
|
destroyClient(client);
|
|
QElapsedTimer t; t.start();
|
|
while (t.elapsed() < 3500) QCoreApplication::processEvents(QEventLoop::AllEvents, 50);
|
|
}
|
|
|
|
TEST_F(CallErrorAfterAcquireTest, SyncModuleNotLoadedOnALiveHostReportsTheError)
|
|
{
|
|
LiveHost host;
|
|
ASSERT_TRUE(host.ok());
|
|
|
|
lp_client* client = clientFor("ghost_module", host.target());
|
|
ASSERT_NE(client, nullptr);
|
|
|
|
char* result = nullptr;
|
|
char* error = nullptr;
|
|
const int rc = lp_invoke(client, "compute", "[]", 5000, &result, &error);
|
|
|
|
std::cout << " SYNC not-published -> rc=" << rc
|
|
<< " result=" << (result ? result : "(null)")
|
|
<< " error=" << (error ? error : "(null)") << std::endl;
|
|
|
|
EXPECT_EQ(rc, LP_ERR_UNAVAILABLE);
|
|
ASSERT_NE(error, nullptr);
|
|
nlohmann::json e = nlohmann::json::parse(error, nullptr, false);
|
|
ASSERT_TRUE(e.is_object());
|
|
EXPECT_EQ(e.value("code", std::string{}), "object_unavailable");
|
|
|
|
lp_string_free(result);
|
|
lp_string_free(error);
|
|
destroyClient(client);
|
|
}
|
|
|
|
// ── the residual gap, pinned deliberately ───────────────────────────────────
|
|
//
|
|
// An UNKNOWN METHOD is NOT fixed here and cannot be at this layer: every
|
|
// provider flavour answers a bare null for a name it doesn't recognise
|
|
// (logos-qt-sdk QtProviderObject's `return QVariant()`, the generated Qt and
|
|
// cdylib dispatches' `unknown method` fall-through, the Rust provider's), which
|
|
// is byte-identical to a method that legitimately returns null. The transport
|
|
// sees ok=true with a null value and MUST report success — reporting failure
|
|
// would break every method whose return really is null. Closing it needs the
|
|
// PROVIDER contract to answer a rejection object for an unknown name, in every
|
|
// SDK, and mirrored into lp_invoke so the twins stay identical.
|
|
//
|
|
// This test asserts today's behaviour so the boundary is explicit rather than
|
|
// assumed, and so it fails loudly if a provider ever starts distinguishing.
|
|
TEST_F(CallErrorAfterAcquireTest, UnknownMethodStaysIndistinguishableFromANullReturn)
|
|
{
|
|
LiveHost host;
|
|
ASSERT_TRUE(host.ok());
|
|
|
|
lp_client* client = clientFor("slow_module", host.target());
|
|
ASSERT_NE(client, nullptr);
|
|
|
|
Capture c;
|
|
ASSERT_EQ(lp_invoke_async(client, "noSuchMethod", "[]", 5000, &captureCb, &c), LP_OK);
|
|
ASSERT_TRUE(pumpUntilFired(c, 15000)) << "async callback never fired";
|
|
|
|
std::cout << " ASYNC unknown method -> ok=" << c.ok << " json=" << c.json
|
|
<< " (residual gap: the provider itself answers a bare null)"
|
|
<< std::endl;
|
|
|
|
EXPECT_EQ(c.ok, 1);
|
|
EXPECT_EQ(c.json, "null");
|
|
|
|
destroyClient(client);
|
|
}
|