diff --git a/src/app.cpp b/src/app.cpp index 42d6911..5597ed3 100644 --- a/src/app.cpp +++ b/src/app.cpp @@ -516,6 +516,9 @@ bool App::init() // Initialize background RPC worker thread worker_ = std::make_unique(); worker_->start(); + // HTTP lane for the market data (price / tickers / charts). Both variants: the lite build has a Market tab. + http_worker_ = std::make_unique(); + http_worker_->start(); network_refresh_.markDue(services::NetworkRefreshService::Timer::Price); // Warm the Mining tab's data at startup (both async / off the UI thread) so pool @@ -827,6 +830,9 @@ void App::update() if (worker_) { worker_->drainResults(); } + if (http_worker_) { + http_worker_->drainResults(); // market data (update() also runs while minimized) + } if (fast_worker_) { fast_worker_->drainResults(); } @@ -6235,6 +6241,13 @@ void App::beginShutdown() if (fast_worker_) { fast_worker_->requestStop(); } + // HTTP lane: requestStop() alone would still run every queued GET to completion (RPCWorker::run drains its + // queue). The flag makes queued GETs return at once and aborts an in-flight transfer through libcurl's + // progress callback; shutdown() joins. + http_cancel_.store(true, std::memory_order_relaxed); + if (http_worker_) { + http_worker_->requestStop(); + } // Drain + join the mining-control thread FIRST so no async start/stop job runs while we tear the miner // down here (avoids two threads driving xmrig_manager_ during shutdown). (M-03 cluster) @@ -7352,6 +7365,7 @@ void App::shutdown() // do synchronous shutdown as fallback. if (!shutting_down_) { DEBUG_LOGF("Synchronous shutdown fallback...\n"); + http_cancel_.store(true, std::memory_order_relaxed); // queued / in-flight market GETs end at once async_tasks_.cancelAll(); async_tasks_.joinAll(); if (rpc_) rpc_->requestAbort(); // unblock any in-flight curl before joining @@ -7362,6 +7376,9 @@ void App::shutdown() if (fast_worker_) { fast_worker_->stop(); } + if (http_worker_) { + http_worker_->stop(); + } if (settings_) { settings_->save(); } @@ -7389,6 +7406,9 @@ void App::shutdown() if (fast_worker_) { fast_worker_->stop(); } + if (http_worker_) { + http_worker_->stop(); // http_cancel_ was set in beginShutdown(): an in-flight GET aborts within ~1 s + } // Disconnect RPC after worker is fully stopped (safe — no curl in flight) if (rpc_) { rpc_->disconnect(); @@ -7459,7 +7479,8 @@ void App::verifyDebugCredential(const std::string& secret, std::functionhasPendingResults()) - || (fast_worker_ && fast_worker_->hasPendingResults()); + || (fast_worker_ && fast_worker_->hasPendingResults()) + || (http_worker_ && http_worker_->hasPendingResults()); } std::string App::transactionSendProgressText() const diff --git a/src/app.h b/src/app.h index 87f29e5..d07b577 100644 --- a/src/app.h +++ b/src/app.h @@ -762,9 +762,9 @@ public: void verifyDebugCredential(const std::string& secret, std::function cb); /// @brief Something finished on a background thread and waits for update() to apply it on the main thread - /// (the RPC worker lanes). main.cpp's idle loop polls this every <= 200 ms (hasImmediateWork) and renders a - /// frame when it is true, so the result shows without input, even while the window is hidden. A new lane whose - /// results are applied in update() adds its term HERE. + /// (the RPC worker lanes, the HTTP lane). main.cpp's idle loop polls this every <= 200 ms (hasImmediateWork) and + /// renders a frame when it is true, so the result shows without input, even while the window is hidden. A new + /// lane whose results are applied in update() adds its term HERE. bool hasPendingBackgroundResults() const; bool hasTransactionSendProgress() const { return send_progress_active_ || send_submissions_in_flight_ > 0 || !pending_opids_.empty(); } std::string transactionSendProgressText() const; @@ -896,6 +896,15 @@ private: std::unique_ptr fast_rpc_; std::unique_ptr fast_worker_; + // HTTP lane: price / tickers / market chart / exchange candle GETs on their own thread, so a slow or + // blocked HTTP endpoint never queues in front of wallet RPCs on worker_ (and wallet scans never delay the + // price). Not tied to the daemon connection: started in init(), stopped only at shutdown. Work functions + // touch only http_cancel_ and their captures; results apply in MainCbs (main thread). http_cancel_ is + // declared BEFORE http_worker_ on purpose: members die in reverse order, and the lane's thread reads the + // flag until ~RPCWorker joins it. + std::atomic http_cancel_{false}; // set at shutdown: queued GETs return at once, an in-flight one aborts + std::unique_ptr http_worker_; + // Saved connection credentials (needed to open the fast-lane connection) rpc::ConnectionConfig saved_config_; @@ -1154,6 +1163,7 @@ private: std::string exchange_chart_fail_key_; std::chrono::steady_clock::time_point market_manual_retry_at_{}; bool price_fetch_failed_ = false; // last price fetch failed (connectivity probe for the backoffs) + bool price_fetch_in_flight_ = false; // a price GET is on http_worker_ (main thread only) // Per-exchange candle chart (refreshExchangeChart): which pair the loaded series is for, an // in-flight guard, and a slow refresh timer (candles move slowly, like the aggregate chart). std::string exchange_chart_key_; // ":/" of the loaded series ("" = none) diff --git a/src/app_network.cpp b/src/app_network.cpp index 9781e56..b21d10e 100644 --- a/src/app_network.cpp +++ b/src/app_network.cpp @@ -3063,52 +3063,41 @@ void App::refreshPeerInfo() if (!enqueued.enqueued) return; } +namespace { +// Market-data GET options for the HTTP lane: a short connect timeout (a blocked endpoint must not hold the +// lane for 15 s), a bounded total, and abortable at shutdown through App::http_cancel_. Deliberate change: +// the CoinGecko chart and exchange candle GETs used the updaters' 15 s / 30 s. +util::HttpGetOptions marketHttp(const std::atomic* cancel, long totalSec) +{ + util::HttpGetOptions o; + o.connectTimeoutSec = 5; + o.timeoutSec = totalSec; + o.cancel = cancel; + return o; +} +} // namespace + void App::refreshPrice() { // Skip if price fetching is disabled if (!settings_->getFetchPrices()) return; - if (!worker_) return; - - auto enqueued = network_refresh_.enqueue(services::NetworkRefreshService::Job::Price, *worker_, [this]() -> rpc::RPCWorker::MainCb { - // --- Worker thread: blocking HTTP GET to CoinGecko --- + if (!http_worker_ || http_cancel_.load(std::memory_order_relaxed)) return; + // A plain in-flight flag, not NetworkRefreshService::enqueue(Job::Price): that one is generation-gated, and + // the resetJobs() of every disconnect would discard a good in-flight price (leaving price_loading set until + // the next tick) for a fetch that has nothing to do with the daemon connection. + if (price_fetch_in_flight_) return; + price_fetch_in_flight_ = true; + NetworkRefreshService::markPriceRefreshStarted(state_); + + http_worker_->post([this]() -> rpc::RPCWorker::MainCb { + // --- HTTP lane: blocking GET to CoinGecko. Touches only http_cancel_ and locals. --- NetworkRefreshService::PriceHttpResult result; - try { - CURL* curl = curl_easy_init(); - if (!curl) { - DEBUG_LOGF("Failed to initialize curl for price fetch\n"); - result.errorMessage = "Price fetch failed: failed to initialize curl"; - } else { - std::string response_data; - const char* url = "https://api.coingecko.com/api/v3/simple/price?ids=dragonx-2&vs_currencies=usd,btc&include_24hr_change=true&include_24hr_vol=true&include_market_cap=true"; - - auto write_callback = [](void* contents, size_t size, size_t nmemb, std::string* userp) -> size_t { - size_t totalSize = size * nmemb; - userp->append((char*)contents, totalSize); - return totalSize; - }; - - curl_easy_setopt(curl, CURLOPT_URL, url); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, +write_callback); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response_data); - curl_easy_setopt(curl, CURLOPT_TIMEOUT, 10L); - curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 5L); - curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); - curl_easy_setopt(curl, CURLOPT_USERAGENT, "DragonX-Wallet/1.0"); - - CURLcode res = curl_easy_perform(curl); - long http_code = 0; - curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code); - curl_easy_cleanup(curl); - - NetworkRefreshService::PriceHttpResponse response{ - res == CURLE_OK, - http_code, - std::move(response_data), - res == CURLE_OK ? std::string() : std::string(curl_easy_strerror(res)) - }; - result = NetworkRefreshService::parsePriceHttpResponse(response, std::time(nullptr)); - } + const char* const url = "https://api.coingecko.com/api/v3/simple/price?ids=dragonx-2&vs_currencies=usd,btc&include_24hr_change=true&include_24hr_vol=true&include_market_cap=true"; + util::HttpGetResult r = util::httpGet(url, "[price]", marketHttp(&http_cancel_, 10)); + NetworkRefreshService::PriceHttpResponse response{ + r.transportOk, r.httpStatus, std::move(r.body), r.error}; + result = NetworkRefreshService::parsePriceHttpResponse(response, std::time(nullptr)); } catch (const std::exception& e) { DEBUG_LOGF("Price fetch error: %s\n", e.what()); result.errorMessage = std::string("Price fetch error: ") + e.what(); @@ -3123,7 +3112,9 @@ void App::refreshPrice() DEBUG_LOGF("%s\n", result.errorMessage.c_str()); } + // Always hand back a callback: it is what clears price_fetch_in_flight_. return [this, result = std::move(result)]() mutable { + price_fetch_in_flight_ = false; if (result.price) { NetworkRefreshService::applyPriceRefreshResult(state_, *result.price, std::chrono::steady_clock::now()); if (price_fetch_failed_) { @@ -3140,9 +3131,7 @@ void App::refreshPrice() price_fetch_failed_ = true; } }; - }, 0); - if (!enqueued.enqueued) return; - NetworkRefreshService::markPriceRefreshStarted(state_); + }); } void App::refreshMarketData() @@ -3167,7 +3156,8 @@ void App::retryMarketDataNow() void App::refreshExchangeChart() { - if (!settings_ || !settings_->getFetchPrices() || !worker_) return; + if (!settings_ || !settings_->getFetchPrices() || !http_worker_ || + http_cancel_.load(std::memory_order_relaxed)) return; if (exchange_chart_fetch_in_flight_) return; // Resolve the selected pair from the same registry the Market tab renders (live tickers, else the @@ -3225,10 +3215,11 @@ void App::refreshExchangeChart() const std::string urlIntra = data::buildExchangeCandleUrl(identifier, base, quote, data::CandleRange::Intraday, now); const std::string urlDaily = data::buildExchangeCandleUrl(identifier, base, quote, data::CandleRange::Daily, now); - worker_->post([this, identifier, key, urlIntra, urlDaily]() -> rpc::RPCWorker::MainCb { - // httpGetString verifies TLS and returns "" on any failure (parses to an empty series). - std::string bIntra = util::httpGetString(urlIntra, "[exch-chart]"); - std::string bDaily = util::httpGetString(urlDaily, "[exch-chart]"); + http_worker_->post([this, identifier, key, urlIntra, urlDaily]() -> rpc::RPCWorker::MainCb { + // httpGetString verifies TLS and returns "" on any failure (parses to an empty series). HTTP lane: + // touches only http_cancel_ and the captures. + std::string bIntra = util::httpGetString(urlIntra, "[exch-chart]", marketHttp(&http_cancel_, 20)); + std::string bDaily = util::httpGetString(urlDaily, "[exch-chart]", marketHttp(&http_cancel_, 20)); auto ohlcIntra = data::parseExchangeOHLC(identifier, bIntra); auto ohlcDaily = data::parseExchangeOHLC(identifier, bDaily); return [this, key, ohlcIntra = std::move(ohlcIntra), ohlcDaily = std::move(ohlcDaily)]() mutable { @@ -3278,7 +3269,7 @@ void App::refreshExchangeChart() void App::refreshMarketChart() { if (!settings_ || !settings_->getFetchPrices()) return; - if (!worker_) return; + if (!http_worker_ || http_cancel_.load(std::memory_order_relaxed)) return; if (chart_fetch_in_flight_) return; const auto now = std::chrono::steady_clock::now(); // Historical data moves slowly — refresh at most every 30 minutes after the first fetch. @@ -3289,17 +3280,17 @@ void App::refreshMarketChart() if (!market_chart_backoff_.due(now)) return; chart_fetch_in_flight_ = true; - worker_->post([this]() -> rpc::RPCWorker::MainCb { + http_worker_->post([this]() -> rpc::RPCWorker::MainCb { // Two ranges give real granularity across every sparkline interval: // days=1 -> 5-minute samples (backs the "hour" interval) // days=365 -> daily samples (backs "day"/"week"/"month") // httpGetString verifies TLS and returns "" on any failure (parses to an empty series). std::string intradayBody = util::httpGetString( "https://api.coingecko.com/api/v3/coins/dragonx-2/market_chart?vs_currency=usd&days=1", - "[market-chart]"); + "[market-chart]", marketHttp(&http_cancel_, 20)); std::string dailyBody = util::httpGetString( "https://api.coingecko.com/api/v3/coins/dragonx-2/market_chart?vs_currency=usd&days=365", - "[market-chart]"); + "[market-chart]", marketHttp(&http_cancel_, 20)); auto intraday = NetworkRefreshService::parseCoinGeckoMarketChart(intradayBody); auto daily = NetworkRefreshService::parseCoinGeckoMarketChart(dailyBody); @@ -3327,19 +3318,18 @@ void App::refreshMarketChart() void App::refreshExchanges() { if (!settings_ || !settings_->getFetchPrices()) return; - if (!worker_) return; + if (!http_worker_ || http_cancel_.load(std::memory_order_relaxed)) return; if (exchanges_fetch_started_) return; // in flight, or loaded (once per session: venues are near-static) // The last attempt failed: wait out the backoff. This is called every frame the Market tab is open, // and a failure used to retry on the very next frame. if (!exchanges_backoff_.due(std::chrono::steady_clock::now())) return; exchanges_fetch_started_ = true; - worker_->post([this]() -> rpc::RPCWorker::MainCb { - // httpGetString verifies TLS, caps the response size and returns "" on any failure. Short - // timeouts: this runs on the shared RPC worker. + http_worker_->post([this]() -> rpc::RPCWorker::MainCb { + // httpGetString verifies TLS, caps the response size and returns "" on any failure. const std::string body = util::httpGetString( "https://api.coingecko.com/api/v3/coins/dragonx-2/tickers", "[tickers]", - /*connectTimeoutSec=*/5, /*timeoutSec=*/12); + marketHttp(&http_cancel_, 12)); std::vector exchanges; if (!body.empty()) { try { exchanges = data::parseCoinGeckoTickers(body); } catch (...) {} diff --git a/src/util/http_download.cpp b/src/util/http_download.cpp index 58164ec..a9b51bb 100644 --- a/src/util/http_download.cpp +++ b/src/util/http_download.cpp @@ -11,6 +11,7 @@ #include #include #include +#include namespace fs = std::filesystem; @@ -47,34 +48,67 @@ int xferCb(void* clientp, curl_off_t dltotal, curl_off_t dlnow, curl_off_t, curl return (*fn)(static_cast(dlnow), static_cast(dltotal)) ? 0 : 1; } +// httpGet's progress callback: abort the transfer (non-zero) once the caller's cancel flag is set. +int cancelXferCb(void* clientp, curl_off_t, curl_off_t, curl_off_t, curl_off_t) +{ + return static_cast*>(clientp)->load(std::memory_order_relaxed) ? 1 : 0; +} + } // namespace +HttpGetResult httpGet(const std::string& url, const char* logTag, const HttpGetOptions& opt) +{ + HttpGetResult out; + if (opt.cancel && opt.cancel->load(std::memory_order_relaxed)) { + out.cancelled = true; + out.error = "cancelled"; + return out; + } + CURL* curl = curl_easy_init(); + if (!curl) { out.error = "failed to initialize curl"; return out; } + curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writeStringCb); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, &out.body); + curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); + curl_easy_setopt(curl, CURLOPT_USERAGENT, "ObsidianDragon/1.0"); + curl_easy_setopt(curl, CURLOPT_TIMEOUT, opt.timeoutSec); + curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, opt.connectTimeoutSec); + curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L); // no SIGALRM for timeouts: these run on worker threads + curl_easy_setopt(curl, CURLOPT_MAXFILESIZE_LARGE, static_cast(kMaxMetadataBytes)); // reject oversized metadata + curl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, 1L); + curl_easy_setopt(curl, CURLOPT_SSL_VERIFYHOST, 2L); + if (opt.cancel) { + curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0L); + curl_easy_setopt(curl, CURLOPT_XFERINFOFUNCTION, cancelXferCb); + curl_easy_setopt(curl, CURLOPT_XFERINFODATA, const_cast*>(opt.cancel)); + } + const CURLcode res = curl_easy_perform(curl); + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &out.httpStatus); + curl_easy_cleanup(curl); + out.transportOk = (res == CURLE_OK); + out.cancelled = (res == CURLE_ABORTED_BY_CALLBACK); + if (!out.transportOk) out.error = out.cancelled ? "cancelled" : curl_easy_strerror(res); + if (!out.cancelled && (!out.transportOk || out.httpStatus < 200 || out.httpStatus >= 300)) { + DEBUG_LOGF("%s GET %s failed: %s (HTTP %ld)\n", + logTag, url.c_str(), curl_easy_strerror(res), out.httpStatus); + } + return out; +} + +std::string httpGetString(const std::string& url, const char* logTag, const HttpGetOptions& opt) +{ + HttpGetResult r = httpGet(url, logTag, opt); + if (!r.transportOk || r.httpStatus < 200 || r.httpStatus >= 300) return {}; + return std::move(r.body); +} + std::string httpGetString(const std::string& url, const char* logTag, long connectTimeoutSec, long timeoutSec) { - CURL* curl = curl_easy_init(); - if (!curl) return {}; - std::string result; - curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writeStringCb); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, &result); - curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); - curl_easy_setopt(curl, CURLOPT_USERAGENT, "ObsidianDragon/1.0"); - curl_easy_setopt(curl, CURLOPT_TIMEOUT, timeoutSec); - curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, connectTimeoutSec); - curl_easy_setopt(curl, CURLOPT_MAXFILESIZE_LARGE, static_cast(kMaxMetadataBytes)); // reject oversized metadata - curl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, 1L); - curl_easy_setopt(curl, CURLOPT_SSL_VERIFYHOST, 2L); - const CURLcode res = curl_easy_perform(curl); - long httpCode = 0; - curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &httpCode); - curl_easy_cleanup(curl); - if (res != CURLE_OK || httpCode < 200 || httpCode >= 300) { - DEBUG_LOGF("%s GET %s failed: %s (HTTP %ld)\n", - logTag, url.c_str(), curl_easy_strerror(res), httpCode); - return {}; - } - return result; + HttpGetOptions opt; + opt.connectTimeoutSec = connectTimeoutSec; + opt.timeoutSec = timeoutSec; + return httpGetString(url, logTag, opt); } bool httpDownloadToFile(const std::string& url, const std::string& destPath, diff --git a/src/util/http_download.h b/src/util/http_download.h index d6dcdba..3acab42 100644 --- a/src/util/http_download.h +++ b/src/util/http_download.h @@ -11,16 +11,38 @@ #pragma once +#include #include #include namespace dragonx { namespace util { -// GET `url` over TLS (peer + host verified) into a string. Follows redirects; 30s transfer / -// 15s connect timeout by default; browser-agnostic User-Agent; responses over 16 MiB are refused. -// Returns "" on ANY failure (curl error or non-2xx status). `logTag` (e.g. "[xmrig-updater]") -// prefixes the debug log on failure. +// Options for httpGet / httpGetString. The defaults are the updaters' (15 s connect, 30 s total). +struct HttpGetOptions { + long connectTimeoutSec = 15; + long timeoutSec = 30; + // When non-null and set (before or during the transfer) the GET is abandoned: checked once up front, + // then by libcurl's progress callback (about once a second even while resolving / connecting). + const std::atomic* cancel = nullptr; +}; + +struct HttpGetResult { + bool transportOk = false; // curl finished without error (any HTTP status) + long httpStatus = 0; // 0 when no response arrived + bool cancelled = false; // abandoned through HttpGetOptions::cancel + std::string body; // whatever arrived (also for a non-2xx status) + std::string error; // curl_easy_strerror text, "cancelled", or "" when transportOk +}; + +// GET `url` over TLS (peer + host verified). Follows redirects; browser-agnostic User-Agent; responses +// over 16 MiB are refused; CURLOPT_NOSIGNAL is set (timeouts are safe on any thread). `logTag` (e.g. +// "[xmrig-updater]") prefixes the debug log on failure. The result says what happened. +HttpGetResult httpGet(const std::string& url, const char* logTag, const HttpGetOptions& opt = {}); + +// The body of a 2xx reply, or "" on ANY failure (curl error, non-2xx status, cancelled). +std::string httpGetString(const std::string& url, const char* logTag, const HttpGetOptions& opt); +// Same, with the updaters' 15 s connect / 30 s total timeouts by default (unchanged contract). std::string httpGetString(const std::string& url, const char* logTag, long connectTimeoutSec = 15, long timeoutSec = 30); diff --git a/tests/test_phase4.cpp b/tests/test_phase4.cpp index d04fd59..4b5578d 100644 --- a/tests/test_phase4.cpp +++ b/tests/test_phase4.cpp @@ -69,6 +69,7 @@ #include "util/xmrig_updater.h" #include "util/pool_registry.h" #include "util/daemon_updater.h" +#include "util/http_download.h" #include "util/lite_server_probe.h" #include "util/wallet_file_probe.h" #include "wallet/lite_connection_service.h" @@ -4431,6 +4432,21 @@ void testFetchBackoff() } } +// C73: a cancelled GET returns before touching the network. +void testHttpGetCancel() +{ + std::atomic cancel{true}; + dragonx::util::HttpGetOptions o; + o.cancel = &cancel; + const dragonx::util::HttpGetResult r = dragonx::util::httpGet("https://127.0.0.1:9/never", "[test]", o); + EXPECT_TRUE(r.cancelled); + EXPECT_FALSE(r.transportOk); + EXPECT_EQ(r.httpStatus, 0L); + EXPECT_TRUE(r.body.empty()); + EXPECT_EQ(r.error, std::string("cancelled")); + EXPECT_EQ(dragonx::util::httpGetString("https://127.0.0.1:9/never", "[test]", o), std::string()); +} + void testConnectRetryBackoff() { using dragonx::rpc::ConnectRetryBackoff; @@ -12792,6 +12808,7 @@ int main() testRpcKindClasses(); testConnectRetryBackoff(); testFetchBackoff(); + testHttpGetCancel(); testNetworkRefreshService(); testNetworkRefreshSnapshotHelpers(); testParseDaemonRescanOutput();