DanS
c1ee089747
The full-node wallet polled the daemon at the per-tab cadence regardless of sync state.
On the Peers/Network tab that meant getpeerinfo every 5s + core every 5s + a full
transaction scan on every new block — and blocks arrive fast during sync. Each of those
calls takes the daemon's cs_main lock, the same lock block connection needs, so the node
synced noticeably slower than on the lightweight Console tab (core 10s, no peer polling).
Make the refresh cadence sync-aware:
- RefreshScheduler::kSyncProfile {core 10s, transactions/addresses/peers disabled} is applied
to ALL tabs while state_.sync.syncing, and reverts to the per-tab profile when sync ends.
applyRefreshPolicy() picks the profile; update() re-applies it on the syncing<->synced
transition. This suppresses getpeerinfo and the per-block tx scan during sync (that data is
incomplete mid-sync anyway) — every tab now syncs as fast as Console.
- collectCoreRefreshResult(rpc, includeBalance): skip z_gettotalbalance (wallet lock + cs_main)
while syncing; only getblockchaininfo runs, which is also what drives sync-progress detection.
applyCoreRefreshResult already leaves the balance untouched when balanceOk is false.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
398 lines
18 KiB
C++
398 lines
18 KiB
C++
#pragma once
|
|
|
|
#include "chat/chat_protocol.h"
|
|
#include "data/wallet_state.h"
|
|
#include "refresh_scheduler.h"
|
|
#include "rpc/rpc_worker.h"
|
|
|
|
#include <nlohmann/json.hpp>
|
|
|
|
#include <array>
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstddef>
|
|
#include <cstdint>
|
|
#include <ctime>
|
|
#include <optional>
|
|
#include <set>
|
|
#include <string>
|
|
#include <unordered_map>
|
|
#include <unordered_set>
|
|
#include <utility>
|
|
#include <vector>
|
|
|
|
namespace dragonx {
|
|
namespace services {
|
|
|
|
class NetworkRefreshService {
|
|
public:
|
|
using Timer = RefreshScheduler::Timer;
|
|
using Intervals = RefreshScheduler::Intervals;
|
|
|
|
class RefreshRpcGateway {
|
|
public:
|
|
virtual ~RefreshRpcGateway() = default;
|
|
virtual nlohmann::json call(const std::string& method,
|
|
const nlohmann::json& params) = 0;
|
|
};
|
|
|
|
enum class Job {
|
|
Core,
|
|
Addresses,
|
|
Transactions,
|
|
Mining,
|
|
Peers,
|
|
Price,
|
|
Encryption,
|
|
ConnectionInit,
|
|
Count
|
|
};
|
|
|
|
struct DispatchTicket {
|
|
Job job = Job::Core;
|
|
std::uint64_t generation = 0;
|
|
bool accepted = false;
|
|
};
|
|
|
|
struct JobStats {
|
|
std::uint64_t started = 0;
|
|
std::uint64_t finished = 0;
|
|
std::uint64_t skippedInFlight = 0;
|
|
std::uint64_t skippedQueuePressure = 0;
|
|
std::uint64_t staleCallbacks = 0;
|
|
std::size_t lastQueueDepth = 0;
|
|
};
|
|
|
|
struct EnqueueResult {
|
|
DispatchTicket ticket;
|
|
bool enqueued = false;
|
|
std::size_t queueDepth = 0;
|
|
};
|
|
|
|
struct ConnectionInfoResult {
|
|
bool ok = false;
|
|
std::optional<int> daemonVersion;
|
|
std::optional<int> protocolVersion;
|
|
std::optional<int> p2pPort;
|
|
std::optional<int> longestChain;
|
|
std::optional<int> notarized;
|
|
std::optional<int> blocks;
|
|
};
|
|
|
|
struct WalletEncryptionResult {
|
|
bool ok = false;
|
|
bool encrypted = false;
|
|
std::int64_t unlockedUntil = 0;
|
|
};
|
|
|
|
struct WarmupPollResult {
|
|
bool ready = false;
|
|
ConnectionInfoResult info;
|
|
std::string errorMessage;
|
|
};
|
|
|
|
struct ConnectionInitResult {
|
|
ConnectionInfoResult info;
|
|
WalletEncryptionResult encryption;
|
|
};
|
|
|
|
struct CoreRefreshResult {
|
|
bool balanceOk = false;
|
|
std::optional<double> shieldedBalance;
|
|
std::optional<double> transparentBalance;
|
|
std::optional<double> totalBalance;
|
|
bool blockchainOk = false;
|
|
std::optional<int> blocks;
|
|
std::optional<int> headers;
|
|
std::optional<std::string> bestBlockHash;
|
|
std::optional<double> verificationProgress;
|
|
std::optional<int> longestChain;
|
|
std::optional<int> notarized;
|
|
};
|
|
|
|
struct MiningRefreshResult {
|
|
std::optional<double> localHashrate;
|
|
bool miningOk = false;
|
|
std::optional<bool> generate;
|
|
std::optional<int> genproclimit;
|
|
std::optional<int> blocks;
|
|
std::optional<double> difficulty;
|
|
std::optional<double> networkHashrate;
|
|
std::optional<std::string> chain;
|
|
double daemonMemoryMb = 0.0;
|
|
};
|
|
|
|
struct PeerRefreshResult {
|
|
std::vector<PeerInfo> peers;
|
|
std::vector<BannedPeer> bannedPeers;
|
|
};
|
|
|
|
struct PriceRefreshResult {
|
|
MarketInfo market;
|
|
};
|
|
|
|
struct PriceHttpResponse {
|
|
bool transportOk = false;
|
|
long httpStatus = 0;
|
|
std::string body;
|
|
std::string transportError;
|
|
};
|
|
|
|
struct PriceHttpResult {
|
|
std::optional<PriceRefreshResult> price;
|
|
std::string errorMessage;
|
|
};
|
|
|
|
struct AddressRefreshResult {
|
|
std::vector<AddressInfo> shieldedAddresses;
|
|
std::vector<AddressInfo> transparentAddresses;
|
|
};
|
|
|
|
struct AddressRefreshSnapshot {
|
|
std::unordered_map<std::string, bool> shieldedSpendingKeys;
|
|
};
|
|
|
|
struct TransactionViewCacheEntry {
|
|
std::string from_address;
|
|
std::int64_t timestamp = 0;
|
|
int confirmations = 0;
|
|
struct Output {
|
|
std::string address;
|
|
double value = 0.0;
|
|
std::string memo;
|
|
std::size_t position = 0;
|
|
};
|
|
std::vector<Output> outgoing_outputs;
|
|
};
|
|
|
|
using TransactionViewCache = std::unordered_map<std::string, TransactionViewCacheEntry>;
|
|
|
|
struct TransactionRefreshSnapshot {
|
|
std::vector<std::string> shieldedAddresses;
|
|
std::unordered_set<std::string> fullyEnrichedTxids;
|
|
TransactionViewCache viewTxCache;
|
|
std::unordered_set<std::string> sendTxids;
|
|
std::unordered_set<std::string> pendingOpids;
|
|
std::vector<TransactionInfo> previousTransactions;
|
|
std::set<std::string> miningAddresses;
|
|
std::unordered_map<std::string, int> shieldedScanHeights;
|
|
std::size_t shieldedScanStartIndex = 0;
|
|
std::size_t maxShieldedReceiveScans = 0;
|
|
// How many blocks the tip may advance past an address's last scan before it counts as stale
|
|
// and needs re-scanning. 0 = strict (must be scanned at the exact current tip). A small
|
|
// tolerance lets a multi-cycle pass over many shielded addresses COMPLETE even though new
|
|
// blocks arrive mid-pass — otherwise the "scanned at tip" bar moves every block and the scan
|
|
// (and the transactions_dirty_ flag it drives) never finishes. It also naturally throttles
|
|
// full rescans to roughly once per `tolerance` blocks.
|
|
int shieldedScanTipTolerance = 0;
|
|
};
|
|
|
|
struct TransactionRefreshResult {
|
|
std::vector<TransactionInfo> transactions;
|
|
std::vector<chat::HushChatTransactionMetadata> hushChatMetadata;
|
|
int blockHeight = -1;
|
|
TransactionViewCache newViewTxEntries;
|
|
std::size_t nextShieldedScanStartIndex = 0;
|
|
std::size_t shieldedAddressesScanned = 0;
|
|
std::size_t shieldedAddressCount = 0;
|
|
std::unordered_map<std::string, int> shieldedScanHeights;
|
|
bool shieldedScanComplete = true;
|
|
};
|
|
|
|
struct OperationStatusPollResult {
|
|
std::vector<std::string> doneOpids;
|
|
std::vector<std::string> staleOpids;
|
|
std::vector<std::string> successTxids;
|
|
std::unordered_map<std::string, std::string> successTxidsByOpid;
|
|
std::vector<std::string> failureMessages;
|
|
std::unordered_map<std::string, std::string> failureByOpid; // opid -> error message
|
|
bool anySuccess = false;
|
|
};
|
|
|
|
struct TransactionCacheUpdate {
|
|
TransactionViewCache& viewTxCache;
|
|
std::unordered_set<std::string>& sendTxids;
|
|
std::vector<TransactionInfo>& confirmedTxCache;
|
|
std::unordered_set<std::string>& confirmedTxIds;
|
|
int& confirmedCacheBlock;
|
|
int& lastTxBlockHeight;
|
|
};
|
|
|
|
static Intervals intervalsForPage(ui::NavPage page) { return RefreshScheduler::intervalsForPage(page); }
|
|
|
|
static ConnectionInfoResult parseConnectionInfoResult(const nlohmann::json& info);
|
|
static WalletEncryptionResult parseWalletEncryptionResult(const nlohmann::json& walletInfo);
|
|
static WarmupPollResult collectWarmupPollResult(RefreshRpcGateway& rpc);
|
|
static ConnectionInitResult collectConnectionInitResult(
|
|
RefreshRpcGateway& rpc,
|
|
const std::optional<ConnectionInfoResult>& prefetchedInfo = std::nullopt);
|
|
static CoreRefreshResult parseCoreRefreshResult(const nlohmann::json& totalBalance,
|
|
bool balanceOk,
|
|
const nlohmann::json& blockInfo,
|
|
bool blockOk);
|
|
// includeBalance=false skips z_gettotalbalance (which takes the wallet lock + cs_main) and only
|
|
// fetches getblockchaininfo — used while syncing, where the balance is incomplete anyway and the
|
|
// wallet should minimise lock contention with block connection.
|
|
static CoreRefreshResult collectCoreRefreshResult(RefreshRpcGateway& rpc, bool includeBalance = true);
|
|
static MiningRefreshResult parseMiningRefreshResult(const nlohmann::json& miningInfo,
|
|
bool miningOk,
|
|
const nlohmann::json& localHashrate,
|
|
bool hashrateOk,
|
|
double daemonMemoryMb);
|
|
static MiningRefreshResult collectMiningRefreshResult(RefreshRpcGateway& rpc,
|
|
double daemonMemoryMb,
|
|
bool includeSlowRefresh,
|
|
bool includeLocalHashrate = true);
|
|
static PeerRefreshResult parsePeerRefreshResult(const nlohmann::json& peers,
|
|
const nlohmann::json& bannedPeers);
|
|
static PeerRefreshResult collectPeerRefreshResult(RefreshRpcGateway& rpc);
|
|
static std::optional<PriceRefreshResult> parseCoinGeckoPriceResponse(const std::string& response,
|
|
std::time_t fetchedAt);
|
|
static PriceHttpResult parsePriceHttpResponse(const PriceHttpResponse& response,
|
|
std::time_t fetchedAt);
|
|
static AddressInfo buildShieldedAddressInfo(const std::string& address,
|
|
const nlohmann::json& validation,
|
|
bool validationSucceeded);
|
|
static AddressInfo buildTransparentAddressInfo(const std::string& address);
|
|
static std::vector<AddressInfo> parseTransparentAddressList(const nlohmann::json& addressList);
|
|
static void applyShieldedBalancesFromUnspent(std::vector<AddressInfo>& addresses,
|
|
const nlohmann::json& unspent);
|
|
static void applyTransparentBalancesFromUnspent(std::vector<AddressInfo>& addresses,
|
|
const nlohmann::json& unspent);
|
|
static AddressRefreshSnapshot buildAddressRefreshSnapshot(const WalletState& state);
|
|
static AddressRefreshResult collectAddressRefreshResult(
|
|
RefreshRpcGateway& rpc,
|
|
const AddressRefreshSnapshot& snapshot = {});
|
|
static TransactionRefreshSnapshot buildTransactionRefreshSnapshot(const WalletState& state,
|
|
const TransactionViewCache& viewTxCache,
|
|
const std::unordered_set<std::string>& sendTxids);
|
|
static void appendTransparentTransactions(std::vector<TransactionInfo>& transactions,
|
|
std::set<std::string>& knownTxids,
|
|
const nlohmann::json& result,
|
|
const std::set<std::string>& miningAddresses = {});
|
|
static void appendShieldedReceivedTransactions(std::vector<TransactionInfo>& transactions,
|
|
std::set<std::string>& knownTxids,
|
|
const std::string& address,
|
|
const nlohmann::json& received,
|
|
const std::set<std::string>& miningAddresses = {},
|
|
std::unordered_map<std::string, std::vector<chat::HushChatMemoOutput>>* chatMemoOutputs = nullptr);
|
|
static TransactionViewCacheEntry parseViewTransactionCacheEntry(const nlohmann::json& viewTransaction);
|
|
static void appendViewTransactionOutputs(std::vector<TransactionInfo>& transactions,
|
|
const std::string& txid,
|
|
const TransactionViewCacheEntry& entry);
|
|
static void sortTransactionsNewestFirst(std::vector<TransactionInfo>& transactions);
|
|
static TransactionRefreshResult collectTransactionRefreshResult(RefreshRpcGateway& rpc,
|
|
const TransactionRefreshSnapshot& snapshot,
|
|
int currentBlockHeight,
|
|
int maxViewTransactionsPerCycle);
|
|
static TransactionRefreshResult collectRecentTransactionRefreshResult(
|
|
RefreshRpcGateway& rpc,
|
|
const TransactionRefreshSnapshot& snapshot,
|
|
int currentBlockHeight,
|
|
int pageSize = 100);
|
|
static OperationStatusPollResult parseOperationStatusPoll(const nlohmann::json& result,
|
|
const std::vector<std::string>& requestedOpids);
|
|
|
|
static void applyConnectionInfoResult(WalletState& state, const ConnectionInfoResult& result);
|
|
static void applyWalletEncryptionResult(WalletState& state, const WalletEncryptionResult& result);
|
|
static void applyConnectionInitResult(WalletState& state, const ConnectionInitResult& result);
|
|
static void applyCoreRefreshResult(WalletState& state,
|
|
const CoreRefreshResult& result,
|
|
std::time_t updatedAt);
|
|
static void applyMiningRefreshResult(WalletState& state,
|
|
const MiningRefreshResult& result,
|
|
std::time_t updatedAt);
|
|
static void applyPeerRefreshResult(WalletState& state,
|
|
PeerRefreshResult&& result,
|
|
std::time_t updatedAt);
|
|
static void markPriceRefreshStarted(WalletState& state);
|
|
static void applyPriceRefreshResult(WalletState& state,
|
|
const PriceRefreshResult& result,
|
|
std::chrono::steady_clock::time_point fetchedAt);
|
|
static void applyPriceRefreshFailure(WalletState& state,
|
|
const std::string& errorMessage);
|
|
static void applyAddressRefreshResult(WalletState& state,
|
|
AddressRefreshResult&& result);
|
|
static void applyTransactionRefreshResult(WalletState& state,
|
|
TransactionCacheUpdate cacheUpdate,
|
|
TransactionRefreshResult&& result,
|
|
std::time_t updatedAt);
|
|
|
|
void applyPage(ui::NavPage page) { scheduler_.applyPage(page); }
|
|
void setIntervals(Intervals intervals) { scheduler_.setIntervals(intervals); }
|
|
const Intervals& intervals() const { return scheduler_.intervals(); }
|
|
|
|
void tick(float deltaSeconds) { scheduler_.tick(deltaSeconds); }
|
|
bool isDue(Timer timer) const { return scheduler_.isDue(timer); }
|
|
bool consumeDue(Timer timer) { return scheduler_.consumeDue(timer); }
|
|
void reset(Timer timer) { scheduler_.reset(timer); }
|
|
void markDue(Timer timer) { scheduler_.markDue(timer); }
|
|
void setTimer(Timer timer, float seconds) { scheduler_.setTimer(timer, seconds); }
|
|
float timer(Timer timer) const { return scheduler_.timer(timer); }
|
|
float interval(Timer timer) const { return scheduler_.interval(timer); }
|
|
void markImmediateRefresh() { scheduler_.markImmediateRefresh(); }
|
|
void markWalletMutationRefresh() { scheduler_.markWalletMutationRefresh(); }
|
|
void resetTxAge() { scheduler_.resetTxAge(); }
|
|
bool shouldRefreshTransactions(int lastTxBlockHeight,
|
|
int currentBlockHeight,
|
|
bool transactionsDirty) const;
|
|
|
|
bool beginJob(Job job);
|
|
bool beginJob(Job job, std::size_t queuedWork, std::size_t maxQueuedWork);
|
|
void finishJob(Job job);
|
|
bool jobInProgress(Job job) const;
|
|
void resetJobs();
|
|
DispatchTicket beginDispatch(Job job, std::size_t queuedWork = 0, std::size_t maxQueuedWork = 0);
|
|
bool completeDispatch(const DispatchTicket& ticket);
|
|
void cancelDispatch(const DispatchTicket& ticket);
|
|
JobStats stats(Job job) const;
|
|
|
|
template <typename Worker, typename WorkFn>
|
|
EnqueueResult enqueue(Job job, Worker& worker, WorkFn&& work, std::size_t maxQueuedWork = 0)
|
|
{
|
|
std::size_t queueDepth = worker.pendingTaskCount();
|
|
auto ticket = beginDispatch(job, queueDepth, maxQueuedWork);
|
|
if (!ticket.accepted) return {ticket, false, queueDepth};
|
|
|
|
worker.post([this, ticket, work = std::forward<WorkFn>(work)]() mutable -> rpc::RPCWorker::MainCb {
|
|
rpc::RPCWorker::MainCb mainCallback;
|
|
try {
|
|
mainCallback = work();
|
|
} catch (...) {
|
|
mainCallback = nullptr;
|
|
}
|
|
return [this, ticket, mainCallback = std::move(mainCallback)]() mutable {
|
|
if (!completeDispatch(ticket)) return;
|
|
if (mainCallback) mainCallback();
|
|
};
|
|
});
|
|
|
|
return {ticket, true, queueDepth};
|
|
}
|
|
|
|
private:
|
|
std::atomic<bool>& jobFlag(Job job);
|
|
const std::atomic<bool>& jobFlag(Job job) const;
|
|
static std::size_t jobIndex(Job job);
|
|
|
|
RefreshScheduler scheduler_;
|
|
std::atomic<bool> coreInProgress_{false};
|
|
std::atomic<bool> addressesInProgress_{false};
|
|
std::atomic<bool> transactionsInProgress_{false};
|
|
std::atomic<bool> miningInProgress_{false};
|
|
std::atomic<bool> peersInProgress_{false};
|
|
std::atomic<bool> priceInProgress_{false};
|
|
std::atomic<bool> encryptionInProgress_{false};
|
|
std::atomic<bool> connectionInitInProgress_{false};
|
|
std::array<std::atomic<std::uint64_t>, static_cast<std::size_t>(Job::Count)> generations_{};
|
|
std::array<std::atomic<std::uint64_t>, static_cast<std::size_t>(Job::Count)> started_{};
|
|
std::array<std::atomic<std::uint64_t>, static_cast<std::size_t>(Job::Count)> finished_{};
|
|
std::array<std::atomic<std::uint64_t>, static_cast<std::size_t>(Job::Count)> skippedInFlight_{};
|
|
std::array<std::atomic<std::uint64_t>, static_cast<std::size_t>(Job::Count)> skippedQueuePressure_{};
|
|
std::array<std::atomic<std::uint64_t>, static_cast<std::size_t>(Job::Count)> staleCallbacks_{};
|
|
std::array<std::atomic<std::size_t>, static_cast<std::size_t>(Job::Count)> lastQueueDepth_{};
|
|
};
|
|
|
|
} // namespace services
|
|
} // namespace dragonx
|