#include "logos_api_client.h" #include "logos_api_consumer.h" #include "logos_object.h" #include "logos_types.h" #include "logos_json_convert.h" #include "logos_thread_marshal.h" #include "logos_rpc_status.h" #include "token_manager.h" #include #include #include #include #include #include #include #include using logos::qvariantToNlohmann; using logos::nlohmannArgsToQVariantList; namespace { // The store this client presents tokens from. // // An explicit store always wins — that is how every host constructs a client // today, and how a host opts an identity in (by passing // &TokenManager::forIdentity(origin) rather than &TokenManager::instance()). // // A NULL store resolves to the origin's store instead of being a guaranteed // crash on the first getToken(). This is the one construction path that cannot // be given an explicit store by its caller — see lp_client_create's frozen // signature — so "no store named" has to mean something, and the only defensible // meaning is "the store for the identity I said I am". TokenManager* storeFor(TokenManager* explicit_store, const QString& origin_module) { return explicit_store ? explicit_store : &TokenManager::forIdentity(origin_module); } } // namespace LogosAPIClient::LogosAPIClient(const QString& module_to_talk_to, const QString& origin_module, TokenManager* token_manager, const LogosTransportConfig& target_transport, const LogosTransportConfig& capability_transport, QObject *parent) : QObject(parent) , m_consumer(new LogosAPIConsumer(module_to_talk_to, origin_module, storeFor(token_manager, origin_module), target_transport, this)) , m_token_manager(storeFor(token_manager, origin_module)) , m_origin_module(origin_module) // Pre-build the capability_module consumer once. We skip it for // the capability_module client itself — the auto-`requestModule` // path is gated by `objectName != "capability_module"` so we'd // never use it, and constructing one would be a redundant // self-connection. Init-list order matches the declaration order // in the header — `m_capability_consumer` is appended at the end // for ABI stability (see header comment). , m_capability_consumer(module_to_talk_to == QStringLiteral("capability_module") ? nullptr : new LogosAPIConsumer(QStringLiteral("capability_module"), origin_module, storeFor(token_manager, origin_module), capability_transport, this)) { } LogosAPIClient::LogosAPIClient(const QString& module_to_talk_to, const QString& origin_module, TokenManager* token_manager, QObject *parent) : LogosAPIClient(module_to_talk_to, origin_module, token_manager, LogosTransportConfigGlobal::getDefault(), LogosTransportConfigGlobal::getDefault(), parent) { } LogosAPIClient::~LogosAPIClient() { } LogosObject* LogosAPIClient::requestObject(const QString& objectName, Timeout timeout) { // Marshal to the owner thread: the replica is acquired and lives there. return logos::runOnOwnerThread(this, [&]() -> LogosObject* { return m_consumer->requestObject(objectName, timeout); }); } bool LogosAPIClient::isConnected() const { return m_consumer->isConnected(); } QString LogosAPIClient::registryUrl() const { return m_consumer->registryUrl(); } bool LogosAPIClient::reconnect() { return m_consumer->reconnect(); } QVariant LogosAPIClient::invokeRemoteMethod(const QString& objectName, const QString& methodName, const QVariantList& args, Timeout timeout) { return invokeRemoteMethod(objectName, methodName, args, timeout, nullptr); } QVariant LogosAPIClient::invokeRemoteMethod(const QString& objectName, const QString& methodName, const QVariantList& args, Timeout timeout, logos::CallError* err) { if (err) err->clear(); // Marshal the whole operation (capability/token fetch + the call) onto the // owner thread so a worker thread (e.g. an HTTP handler) can call other // modules. Same-thread callers run directly. See logos_thread_marshal.h. return logos::runOnOwnerThread(this, [&]() -> QVariant { qDebug() << "LogosAPIClient: invoking remote method" << objectName << methodName << "args_count:" << args.size(); const bool eligible = objectName != QStringLiteral("capability_module") && m_capability_consumer; QString token = getToken(objectName); if (token.isEmpty() && eligible) token = mintAndCacheToken(objectName, timeout); // first exchange (cached for later calls) QVariant result = m_consumer->invokeRemoteMethod(token, objectName, methodName, args, timeout, err); // Re-exchange on rejection, once. The provider rejected our (stale) token — // drop it, mint a fresh one via capability_module, and retry the call. Gated // on the explicit provider sentinel (never a plain empty result), so it can't // loop, can't misfire on a legitimately-empty return, and never fires against // an old provider (which returns a bare QVariant() we don't match). This also // lazily recovers the common provider-reload case. See logos_rpc_status.h. if (eligible && logos::isUnauthorizedSentinel(result)) { qWarning() << "LogosAPIClient: token for" << objectName << "rejected by provider; re-exchanging and retrying once"; m_token_manager->removeToken(objectName); const QString fresh = mintAndCacheToken(objectName, timeout); if (!fresh.isEmpty()) result = m_consumer->invokeRemoteMethod(fresh, objectName, methodName, args, timeout, err); } // Never surface the sentinel to the typed wrapper. If we still hold it the // retry failed (capability down / provider truly gone): collapse to today's // empty result, and for NEW callers set a distinguishable CallError. if (logos::isUnauthorizedSentinel(result)) { if (err) { err->code = "unauthorized"; err->message = "call to '" + objectName.toStdString() + "' rejected: token not recognized (re-exchange failed)"; err->origin = objectName.toStdString(); } return QVariant(); } return result; }); } QString LogosAPIClient::mintAndCacheToken(const QString& objectName, Timeout timeout) { qDebug() << "LogosAPIClient: calling requestModule for" << objectName; const QString capabilityToken = getToken(QStringLiteral("capability_module")); // A NAMED DIAGNOSTIC for the one way this whole path fails silently. // // A private token store is created empty; the host is what puts the // identity's own credential in it. A host that isolates an identity and // never adopts a credential for it produces an empty capability token here, // which ModuleProxy::authorize refuses at its empty-token check — so // requestModule returns "", the real call goes out with no token, the one // re-exchange fails identically, and the caller sees an empty QVariant. That // reads as "the target returned nothing", which is the wrong bug to chase. // // Warned once per client rather than per call: the failure repeats on every // call and the message is about the host's wiring, not about this call. // Only for an ISOLATED store — the ambient ring legitimately starts without // a capability token in plenty of tests and single-module processes. if (capabilityToken.isEmpty() && m_token_manager && m_token_manager != &TokenManager::instance() && !m_warnedNoCredential) { m_warnedNoCredential = true; qWarning() << "LogosAPIClient: identity" << m_origin_module << "has an isolated token store with no credential, so its" " requestModule handshake for" << objectName << "will be refused. The host must admit this identity" " (logos::admitConsumer / TokenManager::adoptCredentialFor)" " before it can call anything."; } const QString token = QString::fromStdString( m_capability_consumer->requestModule(capabilityToken.toStdString(), m_origin_module.toStdString(), objectName.toStdString(), timeout.ms)); qDebug() << "LogosAPIClient: requestModule result for" << objectName << ":" << token; // Cache the minted token so subsequent calls skip the handshake — closes the // token-rotation race where overlapping requestModule calls mint fresh tokens // that overwrite each other at the target (e.g. QtRO's sync wait reentering // via a nested event loop). // // saveToken == the OUTBOUND half, keyed by the CALLEE. It used to be one map // with the inbound tokens, so this write also authorized `objectName` to // call US and an inbound push for the same peer clobbered this cache. Both // are closed; see the DIRECTION note in token_manager.h. if (!token.isEmpty()) m_token_manager->saveToken(objectName, token); return token; } QVariant LogosAPIClient::invokeRemoteMethod(const QString& objectName, const QString& methodName, const QVariant& arg, Timeout timeout) { return invokeRemoteMethod(objectName, methodName, QVariantList() << arg, timeout); } QVariant LogosAPIClient::invokeRemoteMethod(const QString& objectName, const QString& methodName, const QVariant& arg1, const QVariant& arg2, Timeout timeout) { return invokeRemoteMethod(objectName, methodName, QVariantList() << arg1 << arg2, timeout); } QVariant LogosAPIClient::invokeRemoteMethod(const QString& objectName, const QString& methodName, const QVariant& arg1, const QVariant& arg2, const QVariant& arg3, Timeout timeout) { return invokeRemoteMethod(objectName, methodName, QVariantList() << arg1 << arg2 << arg3, timeout); } QVariant LogosAPIClient::invokeRemoteMethod(const QString& objectName, const QString& methodName, const QVariant& arg1, const QVariant& arg2, const QVariant& arg3, const QVariant& arg4, Timeout timeout) { return invokeRemoteMethod(objectName, methodName, QVariantList() << arg1 << arg2 << arg3 << arg4, timeout); } QVariant LogosAPIClient::invokeRemoteMethod(const QString& objectName, const QString& methodName, const QVariant& arg1, const QVariant& arg2, const QVariant& arg3, const QVariant& arg4, const QVariant& arg5, Timeout timeout) { return invokeRemoteMethod(objectName, methodName, QVariantList() << arg1 << arg2 << arg3 << arg4 << arg5, timeout); } void LogosAPIClient::invokeRemoteMethodAsync(const QString& objectName, const QString& methodName, const QVariantList& args, AsyncResultCallback callback, Timeout timeout) { // Delegate to the CallError-aware overload; legacy callers just drop the // error field. Keeps the handshake-coalescing logic single-sourced. invokeRemoteMethodAsync(objectName, methodName, args, [cb = std::move(callback)](QVariant r, const logos::CallError&) mutable { if (cb) cb(std::move(r)); }, timeout); } void LogosAPIClient::invokeRemoteMethodAsync(const QString& objectName, const QString& methodName, const QVariantList& args, AsyncResultErrorCallback callback, Timeout timeout) { // Public entry: grant one retry for the rejection-driven re-exchange. invokeRemoteMethodAsyncImpl(objectName, methodName, args, std::move(callback), timeout, /*retriesLeft=*/1); } void LogosAPIClient::invokeRemoteMethodAsyncImpl(const QString& objectName, const QString& methodName, const QVariantList& args, AsyncResultErrorCallback callback, Timeout timeout, int retriesLeft) { if (!callback) return; // The async path acquires a replica too, so it must also run on the owner // thread. Unlike the sync path we post non-blocking (QueuedConnection): the // worker caller returns immediately and the result callback fires on the // owner thread when the reply arrives. Preserve retriesLeft across the hop. if (QThread::currentThread() != this->thread()) { QMetaObject::invokeMethod(this, [this, objectName, methodName, args, callback = std::move(callback), timeout, retriesLeft]() mutable { invokeRemoteMethodAsyncImpl(objectName, methodName, args, std::move(callback), timeout, retriesLeft); }, Qt::QueuedConnection); return; } const bool eligible = objectName != QStringLiteral("capability_module") && m_capability_consumer; // Wrap the user callback so a provider rejection sentinel triggers one // re-exchange + retry, and the sentinel is never surfaced to the caller. // Mirrors the sync path's retry in logos_api_client.cpp's invokeRemoteMethod. QPointer selfGuard(this); AsyncResultErrorCallback onResult = [this, selfGuard, objectName, methodName, args, timeout, retriesLeft, cb = std::move(callback)] (QVariant result, const logos::CallError& err) mutable { if (!selfGuard) return; // client destroyed mid-flight: drop if (retriesLeft > 0 && objectName != QStringLiteral("capability_module") && m_capability_consumer && logos::isUnauthorizedSentinel(result)) { qWarning() << "LogosAPIClient: token for" << objectName << "rejected by provider (async); re-exchanging and retrying once"; m_token_manager->removeToken(objectName); // Token is empty now → the re-entry coalesces the retry through the // same m_pendingHandshakes machinery, so a burst of concurrent // rejections doesn't restorm capability_module with N handshakes. invokeRemoteMethodAsyncImpl(objectName, methodName, args, std::move(cb), timeout, retriesLeft - 1); return; } if (logos::isUnauthorizedSentinel(result)) { logos::CallError e; e.code = "unauthorized"; e.message = "call to '" + objectName.toStdString() + "' rejected: token not recognized (re-exchange failed)"; e.origin = objectName.toStdString(); cb(QVariant(), e); return; } cb(std::move(result), err); }; QString token = getToken(objectName); if (token.isEmpty() && eligible) { // Async-chain: dispatch the requestModule call asynchronously, and only // fire the real method's invokeRemoteMethodAsync from its callback. The // previous version called `requestModule` synchronously here, which made // the "async" entry point block its caller for the full round-trip. // // COALESCE concurrent first-calls behind ONE handshake. A driver that // fans out N async calls to an un-tokened target before any completes // would otherwise fire N separate requestModule handshakes; each mints a // distinct token and informs the target, and the later inform OVERWRITES // the earlier token there (the target stores one token per caller). The // already-dispatched calls then carry a superseded token and the target // rejects them as unauthorized. So only the first caller starts the // handshake; the rest queue and all drain with the single minted token. // (The sync path can't hit this — it blocks per call, so handshakes // never overlap.) m_pendingHandshakes is touched only on the owner // thread, reached above, so no lock is needed. m_pendingHandshakes[objectName].push_back( [this, objectName, methodName, args, timeout, cb = std::move(onResult)] (const QString& tok) mutable { m_consumer->invokeRemoteMethodAsync(tok, objectName, methodName, args, std::move(cb), timeout); }); if (m_pendingHandshakes[objectName].size() > 1) return; // a handshake for this target is already in flight const QString capabilityToken = getToken("capability_module"); const QString origin = m_origin_module; // Lifetime: capture the client through a QPointer guard. If it (and its // QObject-parented consumers + the pending queue) is destroyed while the // requestModule round-trip is in flight, the guard goes null and we drop // the queued continuations instead of dereferencing dangling memory. QPointer self(this); m_capability_consumer->invokeRemoteMethodAsync( capabilityToken, QStringLiteral("capability_module"), QStringLiteral("requestModule"), QVariantList() << origin << objectName, [self, objectName](const QVariant& tokenResult) mutable { if (!self) return; // client destroyed mid-flight const QString tok = tokenResult.toString(); // Cache the minted token before draining so future calls skip the handshake — m_pendingHandshakes only coalesces the first burst, the cache stops a second burst from racing the same rotation. if (!tok.isEmpty()) self->m_token_manager->saveToken(objectName, tok); // Drain every continuation queued for this target with the one // minted token — the target was informed of exactly this token. // An empty tok (handshake failed) still flows through: the // consumer call is then rejected and each callback fires with an // invalid QVariant, so callers never hang. auto it = self->m_pendingHandshakes.find(objectName); if (it == self->m_pendingHandshakes.end()) return; std::vector> calls = std::move(it.value()); self->m_pendingHandshakes.erase(it); for (auto& c : calls) c(tok); }, timeout); return; } m_consumer->invokeRemoteMethodAsync(token, objectName, methodName, args, std::move(onResult), timeout); } void LogosAPIClient::invokeRemoteMethodAsync(const QString& objectName, const QString& methodName, const QVariant& arg, AsyncResultCallback callback, Timeout timeout) { invokeRemoteMethodAsync(objectName, methodName, QVariantList() << arg, std::move(callback), timeout); } void LogosAPIClient::invokeRemoteMethodAsync(const QString& objectName, const QString& methodName, const QVariant& arg1, const QVariant& arg2, AsyncResultCallback callback, Timeout timeout) { invokeRemoteMethodAsync(objectName, methodName, QVariantList() << arg1 << arg2, std::move(callback), timeout); } void LogosAPIClient::invokeRemoteMethodAsync(const QString& objectName, const QString& methodName, const QVariant& arg1, const QVariant& arg2, const QVariant& arg3, AsyncResultCallback callback, Timeout timeout) { invokeRemoteMethodAsync(objectName, methodName, QVariantList() << arg1 << arg2 << arg3, std::move(callback), timeout); } void LogosAPIClient::invokeRemoteMethodAsync(const QString& objectName, const QString& methodName, const QVariant& arg1, const QVariant& arg2, const QVariant& arg3, const QVariant& arg4, AsyncResultCallback callback, Timeout timeout) { invokeRemoteMethodAsync(objectName, methodName, QVariantList() << arg1 << arg2 << arg3 << arg4, std::move(callback), timeout); } void LogosAPIClient::invokeRemoteMethodAsync(const QString& objectName, const QString& methodName, const QVariant& arg1, const QVariant& arg2, const QVariant& arg3, const QVariant& arg4, const QVariant& arg5, AsyncResultCallback callback, Timeout timeout) { invokeRemoteMethodAsync(objectName, methodName, QVariantList() << arg1 << arg2 << arg3 << arg4 << arg5, std::move(callback), timeout); } void LogosAPIClient::onEvent(LogosObject* originObject, const QString& eventName, std::function callback) { // Marshal to the owner thread: event registration touches the replica. logos::runOnOwnerThread(this, [&]() { m_consumer->onEvent(originObject, eventName, std::move(callback)); }); } quint64 LogosAPIClient::onEventWhenAvailable(const QString& objectName, const QString& eventName, std::function callback, std::function onArmed) { // Marshal to the owner thread for the same reason onEvent() does: the // registry touches (and later arms against) a QtRO replica, which only // works on the thread that created the node. return logos::runOnOwnerThread(this, [&]() -> quint64 { return m_consumer->onEventWhenAvailable(objectName, eventName, std::move(callback), std::move(onArmed)); }); } quint64 LogosAPIClient::whenObjectAvailable(const QString& objectName, std::function onReady) { // Same owner-thread marshalling as onEventWhenAvailable: the registry // touches a QtRO node that only works on the thread that created it. return logos::runOnOwnerThread(this, [&]() -> quint64 { return m_consumer->whenObjectAvailable(objectName, std::move(onReady)); }); } bool LogosAPIClient::cancelEventSubscription(quint64 subscriptionId) { return logos::runOnOwnerThread(this, [&]() -> bool { return m_consumer->cancelEventSubscription(subscriptionId); }); } LogosSubscriptionState LogosAPIClient::eventSubscriptionState(quint64 subscriptionId) const { return logos::runOnOwnerThread(const_cast(this), [&]() -> LogosSubscriptionState { return m_consumer->eventSubscriptionState(subscriptionId); }); } QStringList LogosAPIClient::pendingEventSubscriptions() const { return logos::runOnOwnerThread(const_cast(this), [&]() -> QStringList { return m_consumer->pendingSubscriptions(); }); } void LogosAPIClient::onEventResponse(LogosObject* object, const QString& eventName, const QVariantList& data) { qDebug() << "[LogosObject] LogosAPIClient::onEventResponse" << eventName << "-> LogosObject::emitEvent"; if (eventName.isEmpty()) { qWarning() << "LogosAPIClient: Event name cannot be empty"; return; } if (!object) { qWarning() << "LogosAPIClient: Cannot emit event on null object"; return; } object->emitEvent(eventName, data); } void LogosAPIClient::onEventResponse(QObject* object, const QString& eventName, const QVariantList& data) { qDebug() << "[LogosObject] LogosAPIClient::onEventResponse (QObject* compat)" << eventName; if (eventName.isEmpty()) { qWarning() << "LogosAPIClient: Event name cannot be empty"; return; } if (!object) { qWarning() << "LogosAPIClient: Cannot emit event on null QObject"; return; } QMetaObject::invokeMethod(object, "eventResponse", Qt::DirectConnection, Q_ARG(QString, eventName), Q_ARG(QVariantList, data)); } bool LogosAPIClient::informModuleToken(const QString& authToken, const QString& moduleName, const QString& token) { return m_consumer->informModuleToken(authToken, moduleName, token); } bool LogosAPIClient::informModuleToken(const std::string& authToken, const std::string& moduleName, const std::string& token) { return informModuleToken(QString::fromStdString(authToken), QString::fromStdString(moduleName), QString::fromStdString(token)); } bool LogosAPIClient::informModuleToken_module(const QString& authToken, const QString& originModule, const QString& moduleName, const QString& token, int timeoutMs) { // Marshal to the owner thread, exactly as requestObject/invokeRemoteMethod do. // This path now goes through acquireCachedObject, so it reads and mutates // m_objectCache — declared single-threaded, and holding thread-affine QtRO // handles. Before the handshake surface existed this method used an // uncached requestObject + release(), so it touched no shared state; routing // it onto the cache is what made the missing marshal reachable. // // (LogosAPIClient::informModuleToken — the 3-arg form above — has the same // missing marshal, but it still uses an uncached handle and predates this // change, so it is left alone rather than widened into this fix.) return logos::runOnOwnerThread(this, [&]() -> bool { return m_consumer->informModuleToken_module(authToken, originModule, moduleName, token, timeoutMs); }); } TokenManager* LogosAPIClient::getTokenManager() const { return m_token_manager; } QString LogosAPIClient::getToken(const QString& module_name) { qDebug() << "LogosAPIClient: getToken for module:" << module_name; QString token = m_token_manager->getToken(module_name); if (!token.isEmpty()) { qDebug() << "LogosAPIClient: Found token for module:" << module_name; return token; } qDebug() << "LogosAPIClient: No token found for module:" << module_name; return ""; } // --------------------------------------------------------------------------- // nlohmann::json overloads // --------------------------------------------------------------------------- nlohmann::json LogosAPIClient::invokeRemoteMethod(const std::string& objectName, const std::string& methodName, const nlohmann::json& args, Timeout timeout) { QVariantList qArgs = nlohmannArgsToQVariantList(args); QVariant result = invokeRemoteMethod( QString::fromStdString(objectName), QString::fromStdString(methodName), qArgs, timeout); return qvariantToNlohmann(result); } void LogosAPIClient::onEvent(LogosObject* originObject, const std::string& eventName, std::function callback) { onEvent(originObject, QString::fromStdString(eventName), [cb = std::move(callback)](const QString& name, const QVariantList& data) { nlohmann::json jData = nlohmann::json::array(); for (const QVariant& v : data) jData.push_back(qvariantToNlohmann(v)); cb(name.toStdString(), jData); }); }