DanSandClaude Opus 5.5
71ed92b9f1
Users who keep funds apart under different labelled addresses can now shed notes (smaller wallet, faster scans/witness rebuilds) without the funds being merged into a single address. - Merge/Consolidate dialog: "Where should the funds go?" - Into one address (unchanged) or Keep in each address. The latter lists shielded addresses with 2+ spendable notes (label, note count, amount; too-small ones disabled) and merges each address's notes back into that address with z_mergetoaddress [addr] addr. Addresses with more notes than the batch size get several batches (the daemon locks a batch's notes before returning, so calls take disjoint notes); the dialog suggests a second run after those confirm. Per-address progress; ops keep running if the dialog is closed. The Consolidate (wallet-size) preset opens in this mode; Merge keeps "Into one". Shielded only: a transparent address can't take a merge's output. - services/note_consolidation: groupSpendableNotes, worthConsolidating, estimateBatches, consolidateInPlace (pure; driven by a merge-call function), with unit tests against a stand-in following dragonxd's rules. - fake_dragonxd: optional note wallet (FAKE_WALLET_NOTES, FAKE_MERGE_SECS) with z_mergetoaddress and its operations. - i18n: 17 new keys in all 8 languages; CJK subset rebuilt. - UI sweep: modal-consolidate target. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1527 lines
60 KiB
C++
1527 lines
60 KiB
C++
// fake_dragonxd — a deliberately slow, scriptable stand-in for dragonxd (test tooling only).
|
|
//
|
|
// Reproduces, in minutes and without a 13k-note wallet, the conditions behind the 2026-09 "GUI
|
|
// hard-kills a busy node" incident (see README.md):
|
|
// * dragonxd's HTTP-RPC queue: N workers + a bounded queue (8 + 16 by default, -rpcthreads /
|
|
// -rpcworkqueue honoured). A full queue gets an immediate plain-text HTTP 500 "Work queue depth
|
|
// exceeded" before any handler runs, and a request whose client gave up keeps its slot until a
|
|
// worker reaches it (libevent has no close callback);
|
|
// * a global cs_main that a "witness rebuild" holds for minutes, logging the real markers. Every
|
|
// method except the lock-free `stop` / `help` / `getrescaninfo` takes it;
|
|
// * dragonxd's shutdown: `stop` (or SIGTERM/SIGINT on POSIX; Windows has no signal channel, as in
|
|
// v1.3.0) unbinds the listener at once, answers new requests on open connections with 503, then
|
|
// waits for the rebuild (Phase 2 is not abortable) before "Shutdown: In progress..." -> flush ->
|
|
// "Shutdown: done" -> exit 0.
|
|
// Everything it does is appended to <datadir>/fake_dragonxd.events, so a test can tell a clean exit
|
|
// ("clean-exit") from a hard kill (no "clean-exit"). Knobs are FAKE_* environment variables; see
|
|
// README.md. Built only with -DDRAGONX_BUILD_FAKE_DAEMON=ON and never shipped.
|
|
|
|
#include <nlohmann/json.hpp>
|
|
|
|
#include <algorithm>
|
|
#include <atomic>
|
|
#include <cctype>
|
|
#include <chrono>
|
|
#include <condition_variable>
|
|
#include <csignal>
|
|
#include <cstdarg>
|
|
#include <cstdint>
|
|
#include <cstdio>
|
|
#include <cstdlib>
|
|
#include <cstring>
|
|
#include <ctime>
|
|
#include <deque>
|
|
#include <filesystem>
|
|
#include <fstream>
|
|
#include <map>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <random>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <vector>
|
|
|
|
#ifdef _WIN32
|
|
# ifndef WIN32_LEAN_AND_MEAN
|
|
# define WIN32_LEAN_AND_MEAN
|
|
# endif
|
|
# ifndef NOMINMAX
|
|
# define NOMINMAX
|
|
# endif
|
|
# include <winsock2.h>
|
|
# include <ws2tcpip.h>
|
|
# include <windows.h>
|
|
using socket_t = SOCKET;
|
|
static const socket_t kBadSocket = INVALID_SOCKET;
|
|
static void closeSocket(socket_t s) { closesocket(s); }
|
|
#else
|
|
# include <arpa/inet.h>
|
|
# include <fcntl.h>
|
|
# include <netinet/in.h>
|
|
# include <sys/select.h>
|
|
# include <sys/socket.h>
|
|
# include <unistd.h>
|
|
# ifdef __linux__
|
|
# include <sys/prctl.h>
|
|
# endif
|
|
using socket_t = int;
|
|
static const socket_t kBadSocket = -1;
|
|
static void closeSocket(socket_t s) { ::close(s); }
|
|
#endif
|
|
|
|
#ifndef MSG_NOSIGNAL
|
|
# define MSG_NOSIGNAL 0 // Windows/macOS: SIGPIPE is ignored process-wide instead (POSIX) or absent
|
|
#endif
|
|
|
|
namespace fs = std::filesystem;
|
|
using json = nlohmann::json;
|
|
using Clock = std::chrono::steady_clock;
|
|
using namespace std::chrono_literals;
|
|
|
|
namespace {
|
|
|
|
// =============================================================================================
|
|
// Configuration
|
|
// =============================================================================================
|
|
|
|
double envDouble(const char* name, double def)
|
|
{
|
|
const char* v = std::getenv(name);
|
|
if (!v || !*v) return def;
|
|
char* end = nullptr;
|
|
const double d = std::strtod(v, &end);
|
|
return (end && *end == '\0') ? d : def;
|
|
}
|
|
|
|
long envLong(const char* name, long def) { return static_cast<long>(envDouble(name, static_cast<double>(def))); }
|
|
|
|
struct Config {
|
|
fs::path datadir;
|
|
bool printToConsole = false;
|
|
int rpcPort = 21769;
|
|
int rpcThreads = 8; // DEFAULT_HTTP_THREADS
|
|
int rpcWorkQueue = 16; // DEFAULT_HTTP_WORKQUEUE
|
|
std::string rpcUser, rpcPassword;
|
|
|
|
// FAKE_* knobs. Times are seconds.
|
|
double warmupSecs = 3; // -28 "Loading block index..." before anything else
|
|
double replaySecs = 0; // -28 "Activating best chain..." with UpdateTip lines (startup replay)
|
|
double replayBlocksPerSec = 12;
|
|
double rebuildDelaySecs = 20; // after "Done Loading", so the GUI connects before the rebuild
|
|
double rebuildSecs = 300; // how long the rebuild holds cs_main; 0 = no rebuild
|
|
double phase1Secs = 2; // the abortable, logged part of the rebuild
|
|
double postRebuildLockSecs = 0; // cs_main held after the end line (periodic flush + SetBestChain)
|
|
double flushSecs = 3; // "Shutdown: In progress..." -> "Shutdown: done"
|
|
double updateTipSecs = 36; // block interval once loaded
|
|
double noiseSecs = 20; // an unrelated log line this often (defeats tail heuristics)
|
|
double serverTimeoutSecs = 30; // -rpcservertimeout: idle keep-alive connections are closed
|
|
long prefill = 0; // phantom abandoned requests injected when the rebuild starts
|
|
long height = 3327145;
|
|
long rebuildTotal = 62583;
|
|
long notes = 13632;
|
|
long version = 1030050; // getinfo "version" (real v1.3.0 reports 1030050)
|
|
bool stuckShutdown = false; // the Sep 29 orphan: datadir lock held, no listener, never exits
|
|
bool mining = false; // internal miner on: `stop` must wait for cs_main first
|
|
bool rebuildSilent = false; // hold cs_main like a rebuild but log no rebuild markers at all
|
|
};
|
|
|
|
Config g_cfg;
|
|
|
|
std::map<std::string, std::string> parseArgs(int argc, char** argv)
|
|
{
|
|
std::map<std::string, std::string> out;
|
|
for (int i = 1; i < argc; ++i) {
|
|
std::string a = argv[i];
|
|
const size_t start = a.find_first_not_of('-');
|
|
if (start == 0 || start == std::string::npos) continue;
|
|
a = a.substr(start);
|
|
const size_t eq = a.find('=');
|
|
if (eq == std::string::npos) out[a] = "1";
|
|
else out[a.substr(0, eq)] = a.substr(eq + 1);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
std::map<std::string, std::string> parseConf(const fs::path& path)
|
|
{
|
|
std::map<std::string, std::string> out;
|
|
std::ifstream f(path);
|
|
std::string line;
|
|
while (std::getline(f, line)) {
|
|
if (!line.empty() && line.back() == '\r') line.pop_back();
|
|
const size_t hash = line.find('#');
|
|
if (hash != std::string::npos) line.erase(hash);
|
|
const size_t eq = line.find('=');
|
|
if (eq == std::string::npos) continue;
|
|
auto trim = [](std::string s) {
|
|
const size_t b = s.find_first_not_of(" \t");
|
|
const size_t e = s.find_last_not_of(" \t");
|
|
return b == std::string::npos ? std::string() : s.substr(b, e - b + 1);
|
|
};
|
|
out[trim(line.substr(0, eq))] = trim(line.substr(eq + 1));
|
|
}
|
|
return out;
|
|
}
|
|
|
|
fs::path defaultDatadir()
|
|
{
|
|
#ifdef _WIN32
|
|
const char* appdata = std::getenv("APPDATA");
|
|
return fs::path(appdata ? appdata : ".") / "Hush" / "DRAGONX";
|
|
#else
|
|
const char* home = std::getenv("HOME");
|
|
const fs::path h = home ? home : ".";
|
|
# ifdef __APPLE__
|
|
return h / "Library" / "Application Support" / "Hush" / "DRAGONX";
|
|
# else
|
|
return h / ".hush" / "DRAGONX";
|
|
# endif
|
|
#endif
|
|
}
|
|
|
|
// =============================================================================================
|
|
// Time, logging, events
|
|
// =============================================================================================
|
|
|
|
long long unixMs()
|
|
{
|
|
return std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
std::chrono::system_clock::now().time_since_epoch()).count();
|
|
}
|
|
|
|
std::string utcStamp(std::time_t t)
|
|
{
|
|
std::tm tm{};
|
|
#ifdef _WIN32
|
|
gmtime_s(&tm, &t);
|
|
#else
|
|
gmtime_r(&t, &tm);
|
|
#endif
|
|
char buf[32];
|
|
std::strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M:%S", &tm);
|
|
return buf;
|
|
}
|
|
|
|
std::string vformat(const char* fmt, va_list ap)
|
|
{
|
|
char buf[2048];
|
|
std::vsnprintf(buf, sizeof(buf), fmt, ap);
|
|
return buf;
|
|
}
|
|
|
|
// dragonxd's LogPrintf: with -printtoconsole every line goes to stdout INSTEAD of debug.log and
|
|
// without a timestamp; otherwise to <datadir>/debug.log with a "YYYY-MM-DD HH:MM:SS " UTC prefix.
|
|
class Logger {
|
|
public:
|
|
void open(const fs::path& datadir, bool console)
|
|
{
|
|
console_ = console;
|
|
if (!console_) file_ = std::fopen((datadir / "debug.log").string().c_str(), "ab");
|
|
}
|
|
void line(const std::string& msg)
|
|
{
|
|
std::lock_guard<std::mutex> lk(mx_);
|
|
if (console_) {
|
|
std::fputs((msg + "\n").c_str(), stdout); // EPIPE after the reader left is ignored, like dragonxd
|
|
std::fflush(stdout);
|
|
} else if (file_) {
|
|
std::fprintf(file_, "%s %s\n", utcStamp(std::time(nullptr)).c_str(), msg.c_str());
|
|
std::fflush(file_);
|
|
}
|
|
}
|
|
void close()
|
|
{
|
|
std::lock_guard<std::mutex> lk(mx_);
|
|
if (file_) std::fclose(file_);
|
|
file_ = nullptr;
|
|
}
|
|
|
|
private:
|
|
std::mutex mx_;
|
|
FILE* file_ = nullptr;
|
|
bool console_ = false;
|
|
};
|
|
|
|
Logger g_log;
|
|
|
|
void logf(const char* fmt, ...)
|
|
{
|
|
va_list ap;
|
|
va_start(ap, fmt);
|
|
const std::string s = vformat(fmt, ap);
|
|
va_end(ap);
|
|
g_log.line(s);
|
|
}
|
|
|
|
// <datadir>/fake_dragonxd.events: one "<name> t=<unix ms> [k=v ...]" line per event, for tests.
|
|
std::mutex g_eventsMx;
|
|
FILE* g_events = nullptr;
|
|
|
|
void event(const std::string& name, const std::string& extra = "")
|
|
{
|
|
std::lock_guard<std::mutex> lk(g_eventsMx);
|
|
if (!g_events) return;
|
|
std::fprintf(g_events, "%s t=%lld%s%s\n", name.c_str(), unixMs(), extra.empty() ? "" : " ", extra.c_str());
|
|
std::fflush(g_events);
|
|
}
|
|
|
|
// =============================================================================================
|
|
// Shared daemon state
|
|
// =============================================================================================
|
|
|
|
std::mutex g_csMain; // the daemon's cs_main
|
|
std::atomic<long> g_height{0};
|
|
std::atomic<bool> g_warmup{true};
|
|
std::mutex g_warmupMx;
|
|
std::string g_warmupStatus = "Loading block index...";
|
|
std::atomic<bool> g_minerOn{false};
|
|
|
|
enum ShutdownReason { kNoShutdown = 0, kRpcStop = 1, kSignal = 2 };
|
|
std::atomic<int> g_shutdownReason{kNoShutdown};
|
|
std::atomic<int> g_stopCount{0};
|
|
volatile std::sig_atomic_t g_signalled = 0;
|
|
|
|
extern "C" void onTermSignal(int) { g_signalled = 1; }
|
|
|
|
void requestShutdown(ShutdownReason reason)
|
|
{
|
|
if (reason == kRpcStop) {
|
|
const int n = ++g_stopCount;
|
|
event("stop-received", "n=" + std::to_string(n));
|
|
}
|
|
int expected = kNoShutdown;
|
|
if (g_shutdownReason.compare_exchange_strong(expected, reason) && reason == kSignal) event("sigterm");
|
|
}
|
|
|
|
bool shutdownRequested()
|
|
{
|
|
if (g_signalled && g_shutdownReason.load() == kNoShutdown) requestShutdown(kSignal);
|
|
return g_shutdownReason.load() != kNoShutdown;
|
|
}
|
|
|
|
// Sleeps up to `secs`; returns false (early) once a shutdown is requested.
|
|
bool sleepUnlessShutdown(double secs)
|
|
{
|
|
const auto end = Clock::now() + std::chrono::duration_cast<Clock::duration>(std::chrono::duration<double>(secs));
|
|
while (Clock::now() < end) {
|
|
if (shutdownRequested()) return false;
|
|
std::this_thread::sleep_for(50ms);
|
|
}
|
|
return !shutdownRequested();
|
|
}
|
|
|
|
void setInitMessage(const std::string& msg)
|
|
{
|
|
logf("init message: %s", msg.c_str());
|
|
std::lock_guard<std::mutex> lk(g_warmupMx);
|
|
g_warmupStatus = msg;
|
|
}
|
|
|
|
std::string warmupStatus()
|
|
{
|
|
std::lock_guard<std::mutex> lk(g_warmupMx);
|
|
return g_warmupStatus;
|
|
}
|
|
|
|
struct Stats {
|
|
std::atomic<long> requests{0}, queueFull{0}, unavailable{0}, unhandled{0}, unauthorized{0};
|
|
std::atomic<int> abandonedNow{0}, maxAbandoned{0}, maxSlots{0};
|
|
};
|
|
|
|
Stats g_stats;
|
|
|
|
// Requests per JSON-RPC method, for "what is the GUI polling" (the `methods` event).
|
|
std::mutex g_methodsMx;
|
|
std::map<std::string, long> g_methodCounts;
|
|
|
|
std::string topMethodsLine(size_t n)
|
|
{
|
|
std::vector<std::pair<long, std::string>> v;
|
|
{
|
|
std::lock_guard<std::mutex> lk(g_methodsMx);
|
|
for (const auto& kv : g_methodCounts) v.push_back({kv.second, kv.first});
|
|
}
|
|
std::sort(v.rbegin(), v.rend());
|
|
std::string out;
|
|
for (size_t i = 0; i < v.size() && i < n; ++i) out += (i ? " " : "") + v[i].second + "=" + std::to_string(v[i].first);
|
|
return out;
|
|
}
|
|
|
|
bool raiseMax(std::atomic<int>& max, int value)
|
|
{
|
|
int cur = max.load();
|
|
while (value > cur) {
|
|
if (max.compare_exchange_weak(cur, value)) return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
void noteAbandoned(int delta)
|
|
{
|
|
const int now = (g_stats.abandonedNow += delta);
|
|
if (delta > 0 && raiseMax(g_stats.maxAbandoned, now)) event("max-abandoned", "n=" + std::to_string(now));
|
|
}
|
|
|
|
std::string statsLine()
|
|
{
|
|
return "requests=" + std::to_string(g_stats.requests.load()) +
|
|
" queue_full_500=" + std::to_string(g_stats.queueFull.load()) +
|
|
" unavailable_503=" + std::to_string(g_stats.unavailable.load()) +
|
|
" unhandled_500=" + std::to_string(g_stats.unhandled.load()) +
|
|
" unauthorized_401=" + std::to_string(g_stats.unauthorized.load()) +
|
|
" max_slots=" + std::to_string(g_stats.maxSlots.load()) +
|
|
" max_abandoned=" + std::to_string(g_stats.maxAbandoned.load()) +
|
|
" abandoned_now=" + std::to_string(g_stats.abandonedNow.load()) +
|
|
" stops=" + std::to_string(g_stopCount.load());
|
|
}
|
|
|
|
// =============================================================================================
|
|
// Canned RPC results
|
|
// =============================================================================================
|
|
|
|
std::uint64_t splitmix64(std::uint64_t x)
|
|
{
|
|
x += 0x9E3779B97F4A7C15ULL;
|
|
x = (x ^ (x >> 30)) * 0xBF58476D1CE4E5B9ULL;
|
|
x = (x ^ (x >> 27)) * 0x94D049BB133111EBULL;
|
|
return x ^ (x >> 31);
|
|
}
|
|
|
|
std::string fakeHash(long h)
|
|
{
|
|
char buf[80];
|
|
std::snprintf(buf, sizeof(buf), "%016llx%016llx%016llx%016llx",
|
|
(unsigned long long)splitmix64(h * 4 + 0), (unsigned long long)splitmix64(h * 4 + 1),
|
|
(unsigned long long)splitmix64(h * 4 + 2), (unsigned long long)splitmix64(h * 4 + 3));
|
|
return buf;
|
|
}
|
|
|
|
const char* const kFakeZAddr =
|
|
"zs1fakedragonxd0000000000000000000000000000000000000000000000000000000000000";
|
|
const char* const kFakeTAddr = "RFakeDragonxdTransparent0000000000";
|
|
|
|
struct Canned {
|
|
bool known = true;
|
|
json result;
|
|
int errCode = 0;
|
|
std::string errMsg;
|
|
};
|
|
|
|
// ---------------------------------------------------------------------------------------------
|
|
// Optional note wallet (FAKE_WALLET_NOTES="addr:count:amount,..."): shielded notes per address, plus
|
|
// z_mergetoaddress and its async operations with dragonxd's rules — confirmed, unlocked notes only;
|
|
// a batch's notes are locked when the call returns; "only source == destination with one note" is
|
|
// refused; the merged note appears one block later. Everything here runs under g_csMain (execOne).
|
|
// ---------------------------------------------------------------------------------------------
|
|
struct FakeNote { long id; std::string address; double amount; long height; bool locked; };
|
|
struct FakeOp {
|
|
std::string id;
|
|
std::vector<long> noteIds;
|
|
std::string to;
|
|
double fee = 0.0;
|
|
double finishAt = 0.0; // unixMs()/1000.0
|
|
std::string status = "executing";
|
|
long long created = 0;
|
|
};
|
|
bool g_noteWallet = false;
|
|
std::vector<FakeNote> g_notes;
|
|
std::vector<FakeOp> g_ops;
|
|
long g_noteSeq = 0;
|
|
long g_opSeq = 0;
|
|
double g_mergeSecs = 6.0;
|
|
|
|
void loadNoteWallet()
|
|
{
|
|
const char* spec = std::getenv("FAKE_WALLET_NOTES");
|
|
if (!spec || !*spec) return;
|
|
g_noteWallet = true;
|
|
g_mergeSecs = envDouble("FAKE_MERGE_SECS", 6.0);
|
|
std::string s(spec);
|
|
size_t pos = 0;
|
|
while (pos <= s.size()) {
|
|
size_t comma = s.find(',', pos);
|
|
if (comma == std::string::npos) comma = s.size();
|
|
const std::string item = s.substr(pos, comma - pos);
|
|
const size_t c1 = item.find(':'), c2 = item.find(':', c1 == std::string::npos ? 0 : c1 + 1);
|
|
if (c1 != std::string::npos && c2 != std::string::npos) {
|
|
const std::string addr = item.substr(0, c1);
|
|
const long count = std::strtol(item.c_str() + c1 + 1, nullptr, 10);
|
|
const double amount = std::strtod(item.c_str() + c2 + 1, nullptr);
|
|
for (long i = 0; i < count; ++i)
|
|
g_notes.push_back({++g_noteSeq, addr, amount, g_height.load() - 10 - i, false});
|
|
}
|
|
pos = comma + 1;
|
|
}
|
|
}
|
|
|
|
std::vector<std::string> noteAddresses()
|
|
{
|
|
std::vector<std::string> out;
|
|
for (const auto& n : g_notes)
|
|
if (std::find(out.begin(), out.end(), n.address) == out.end()) out.push_back(n.address);
|
|
return out;
|
|
}
|
|
|
|
// Operations past their run time succeed: their notes are spent, one merged note lands next block.
|
|
void settleOps()
|
|
{
|
|
const double now = unixMs() / 1000.0;
|
|
for (auto& op : g_ops) {
|
|
if (op.status != "executing" || now < op.finishAt) continue;
|
|
double sum = 0.0;
|
|
for (long id : op.noteIds) {
|
|
auto it = std::find_if(g_notes.begin(), g_notes.end(), [&](const FakeNote& n) { return n.id == id; });
|
|
if (it != g_notes.end()) { sum += it->amount; g_notes.erase(it); }
|
|
}
|
|
g_notes.push_back({++g_noteSeq, op.to, sum - op.fee, g_height.load() + 1, false});
|
|
op.status = "success";
|
|
event("merge-done", op.id + " notes=" + std::to_string(op.noteIds.size()) + " to=" + op.to);
|
|
}
|
|
}
|
|
|
|
json opStatus(const FakeOp& op)
|
|
{
|
|
json o = {{"id", op.id}, {"status", op.status}, {"creation_time", op.created}, {"method", "z_mergetoaddress"}};
|
|
if (op.status == "success") o["result"] = {{"txid", fakeHash(1000000 + std::stol(op.id.substr(5)))}};
|
|
return o;
|
|
}
|
|
|
|
// The note-wallet answer for `m`, or c.known = false when it isn't a note-wallet method.
|
|
Canned noteWalletReply(const std::string& m, const json& params)
|
|
{
|
|
Canned c;
|
|
settleOps();
|
|
const long h = g_height.load();
|
|
if (m == "z_listunspent") {
|
|
json arr = json::array();
|
|
for (const auto& n : g_notes) {
|
|
const long conf = std::max(0L, h - n.height + 1);
|
|
arr.push_back({{"txid", fakeHash(5000000 + n.id)}, {"outindex", 0}, {"confirmations", conf},
|
|
{"spendable", true}, {"address", n.address}, {"amount", n.amount}, {"memo", "f6"},
|
|
{"change", false}});
|
|
}
|
|
c.result = arr;
|
|
} else if (m == "z_listaddresses") {
|
|
json arr = json::array({kFakeZAddr});
|
|
for (const auto& a : noteAddresses()) arr.push_back(a);
|
|
c.result = arr;
|
|
} else if (m == "z_gettotalbalance" || m == "z_getbalance") {
|
|
const std::string only = (m == "z_getbalance" && !params.empty() && params[0].is_string())
|
|
? params[0].get<std::string>() : std::string();
|
|
double sum = 0.0;
|
|
for (const auto& n : g_notes)
|
|
if (only.empty() || n.address == only) sum += n.amount;
|
|
if (m == "z_getbalance") c.result = sum;
|
|
else {
|
|
char buf[64];
|
|
std::snprintf(buf, sizeof(buf), "%.8f", sum);
|
|
c.result = {{"transparent", "0.00"}, {"private", buf}, {"total", buf}};
|
|
}
|
|
} else if (m == "z_mergetoaddress") {
|
|
if (params.size() < 2 || !params[0].is_array() || !params[1].is_string()) {
|
|
c.errCode = -8; c.errMsg = "Invalid parameter";
|
|
return c;
|
|
}
|
|
std::vector<std::string> from;
|
|
bool anySapling = false;
|
|
for (const auto& a : params[0]) {
|
|
if (!a.is_string()) continue;
|
|
if (a.get<std::string>() == "ANY_SAPLING" || a.get<std::string>() == "*") anySapling = true;
|
|
else from.push_back(a.get<std::string>());
|
|
}
|
|
const std::string to = params[1].get<std::string>();
|
|
const double fee = params.size() > 2 && params[2].is_number() ? params[2].get<double>() : 0.0001;
|
|
long limit = params.size() > 4 && params[4].is_number() ? params[4].get<long>() : 90;
|
|
if (limit <= 0) limit = 1000000;
|
|
std::vector<long> picked;
|
|
double merging = 0.0, remainingValue = 0.0;
|
|
long counter = 0;
|
|
for (const auto& n : g_notes) {
|
|
if (n.locked || h - n.height + 1 < 1) continue;
|
|
if (!anySapling && std::find(from.begin(), from.end(), n.address) == from.end()) continue;
|
|
++counter;
|
|
if (static_cast<long>(picked.size()) < limit) { picked.push_back(n.id); merging += n.amount; }
|
|
else remainingValue += n.amount;
|
|
}
|
|
if (picked.empty()) { c.errCode = -6; c.errMsg = "Could not find any funds to merge."; return c; }
|
|
if (from.size() == 1 && !anySapling && from[0] == to && picked.size() == 1) {
|
|
c.errCode = -8;
|
|
c.errMsg = "Destination address is also the only source address, and all its funds are already merged.";
|
|
return c;
|
|
}
|
|
if (merging - fee < fee) {
|
|
c.errCode = -6; c.errMsg = "Insufficient funds to pay the fee"; return c;
|
|
}
|
|
for (auto& n : g_notes)
|
|
if (std::find(picked.begin(), picked.end(), n.id) != picked.end()) n.locked = true;
|
|
FakeOp op;
|
|
op.id = "opid-" + std::to_string(++g_opSeq);
|
|
op.noteIds = picked;
|
|
op.to = to;
|
|
op.fee = fee;
|
|
op.created = unixMs() / 1000;
|
|
op.finishAt = unixMs() / 1000.0 + g_mergeSecs;
|
|
g_ops.push_back(op);
|
|
event("merge", op.id + " notes=" + std::to_string(picked.size()) + " to=" + to);
|
|
c.result = {{"remainingUTXOs", 0}, {"remainingTransparentValue", 0.0},
|
|
{"remainingNotes", counter - static_cast<long>(picked.size())},
|
|
{"remainingShieldedValue", remainingValue}, {"mergingUTXOs", 0},
|
|
{"mergingTransparentValue", 0.0}, {"mergingNotes", static_cast<long>(picked.size())},
|
|
{"mergingShieldedValue", merging}, {"opid", op.id}};
|
|
} else if (m == "z_getoperationstatus" || m == "z_getoperationresult") {
|
|
std::vector<std::string> filter;
|
|
if (!params.empty() && params[0].is_array())
|
|
for (const auto& id : params[0]) if (id.is_string()) filter.push_back(id.get<std::string>());
|
|
json arr = json::array();
|
|
std::vector<FakeOp> keep;
|
|
for (const auto& op : g_ops) {
|
|
const bool match = filter.empty() || std::find(filter.begin(), filter.end(), op.id) != filter.end();
|
|
const bool finished = op.status != "executing";
|
|
if (match && (m == "z_getoperationstatus" || finished)) arr.push_back(opStatus(op));
|
|
if (!(m == "z_getoperationresult" && match && finished)) keep.push_back(op);
|
|
}
|
|
g_ops.swap(keep);
|
|
c.result = arr;
|
|
} else {
|
|
c.known = false;
|
|
}
|
|
return c;
|
|
}
|
|
|
|
Canned cannedReply(const std::string& m, const json& params)
|
|
{
|
|
Canned c;
|
|
const long h = g_height.load();
|
|
const long long now = unixMs() / 1000;
|
|
const double difficulty = 152341.734;
|
|
const std::string sub = "/DragonX:" + std::to_string(g_cfg.version / 1000000) + "." +
|
|
std::to_string(g_cfg.version / 10000 % 100) + "." +
|
|
std::to_string(g_cfg.version / 100 % 100) + "/";
|
|
auto param0 = [&]() -> std::string {
|
|
if (params.empty()) return "";
|
|
return params[0].is_string() ? params[0].get<std::string>() : params[0].dump();
|
|
};
|
|
|
|
if (m == "getinfo") {
|
|
c.result = {{"version", g_cfg.version}, {"protocolversion", 170012}, {"walletversion", 60000},
|
|
{"balance", 0.0}, {"blocks", h}, {"longestchain", h}, {"notarized", 0},
|
|
{"tiptime", now}, {"difficulty", difficulty}, {"keypoololdest", now - 86400},
|
|
{"keypoolsize", 101}, {"paytxfee", 0.0}, {"timeoffset", 0}, {"connections", 2},
|
|
{"tls_connections", 2}, {"proxy", ""}, {"testnet", false}, {"relayfee", 0.000001},
|
|
{"errors", ""}, {"name", "DRAGONX"}, {"p2pport", 21768}, {"rpcport", g_cfg.rpcPort}};
|
|
} else if (m == "getblockchaininfo") {
|
|
c.result = {{"chain", "main"}, {"blocks", h}, {"synced", true}, {"headers", h},
|
|
{"bestblockhash", fakeHash(h)}, {"difficulty", difficulty}, {"verificationprogress", 1.0},
|
|
{"chainwork", std::string(48, '0') + "1f3a5c7e9b2d4f60"}, {"pruned", false},
|
|
{"valuePools", json::array()}, {"softforks", json::array()}, {"upgrades", json::object()},
|
|
{"longestchain", h}, {"notarized", 0}};
|
|
} else if (m == "getblockcount") {
|
|
c.result = h;
|
|
} else if (m == "getbestblockhash") {
|
|
c.result = fakeHash(h);
|
|
} else if (m == "getblockhash") {
|
|
c.result = fakeHash(params.empty() || !params[0].is_number() ? h : params[0].get<long>());
|
|
} else if (m == "getblock") {
|
|
long bh = h;
|
|
const std::string p = param0();
|
|
if (!p.empty() && p.size() < 12) bh = std::strtol(p.c_str(), nullptr, 10);
|
|
c.result = {{"hash", fakeHash(bh)}, {"confirmations", std::max(1L, h - bh + 1)}, {"height", bh},
|
|
{"version", 4}, {"time", now - (h - bh) * 36}, {"difficulty", difficulty},
|
|
{"previousblockhash", fakeHash(bh - 1)}, {"tx", json::array({fakeHash(bh + 7777777)})},
|
|
{"size", 1765}};
|
|
} else if (m == "getwalletinfo") {
|
|
c.result = {{"walletversion", 60000}, {"balance", 0.0}, {"unconfirmed_balance", 0.0},
|
|
{"immature_balance", 0.0}, {"txcount", 42}, {"keypoololdest", now - 86400},
|
|
{"keypoolsize", 101}, {"paytxfee", 0.0}};
|
|
} else if (m == "z_gettotalbalance") {
|
|
c.result = {{"transparent", "0.00"}, {"private", "1234.5678"}, {"total", "1234.5678"}};
|
|
} else if (m == "z_getbalance") {
|
|
c.result = 1234.5678;
|
|
} else if (m == "getbalance") {
|
|
c.result = 0.0;
|
|
} else if (m == "getmininginfo") {
|
|
c.result = {{"blocks", h}, {"currentblocksize", 0}, {"currentblocktx", 0}, {"difficulty", difficulty},
|
|
{"errors", ""}, {"genproclimit", -1}, {"localsolps", 0.0}, {"networksolps", 182345},
|
|
{"networkhashps", 182345}, {"pooledtx", 0}, {"testnet", false}, {"chain", "main"},
|
|
{"generate", g_minerOn.load()}, {"numthreads", -1}};
|
|
} else if (m == "getlocalsolps") {
|
|
c.result = 0.0;
|
|
} else if (m == "getnetworkhashps" || m == "getnetworksolps") {
|
|
c.result = 182345;
|
|
} else if (m == "setgenerate") {
|
|
c.result = nullptr; // the miner flag is flipped in execOne: this table stays side-effect free
|
|
} else if (m == "getgenerate") {
|
|
c.result = g_minerOn.load();
|
|
} else if (m == "getpeerinfo") {
|
|
// Documentation-range (TEST-NET) addresses only.
|
|
json peers = json::array();
|
|
const char* addrs[] = {"203.0.113.10:21768", "198.51.100.23:21768"};
|
|
for (int i = 0; i < 2; ++i) {
|
|
peers.push_back({{"id", i + 1}, {"addr", addrs[i]}, {"services", "0000000000000005"},
|
|
{"tls_verified", false}, {"lastsend", now}, {"lastrecv", now},
|
|
{"bytessent", 123456}, {"bytesrecv", 654321}, {"conntime", now - 3600},
|
|
{"timeoffset", 0}, {"pingtime", 0.052}, {"version", 170012}, {"subver", sub},
|
|
{"inbound", false}, {"startingheight", h - 100}, {"banscore", 0},
|
|
{"synced_headers", h}, {"synced_blocks", h}, {"inflight", json::array()},
|
|
{"whitelisted", false}});
|
|
}
|
|
c.result = peers;
|
|
} else if (m == "getnetworkinfo") {
|
|
c.result = {{"version", g_cfg.version}, {"subversion", sub}, {"protocolversion", 170012},
|
|
{"localservices", "0000000000000005"}, {"timeoffset", 0}, {"connections", 2},
|
|
{"tls_connections", 2}, {"networks", json::array()}, {"relayfee", 0.000001},
|
|
{"localaddresses", json::array()}, {"warnings", ""}};
|
|
} else if (m == "getconnectioncount") {
|
|
c.result = 2;
|
|
} else if (m == "getmempoolinfo") {
|
|
c.result = {{"size", 0}, {"bytes", 0}, {"usage", 0}};
|
|
} else if (m == "listbanned" || m == "listunspent" || m == "z_listunspent" || m == "listtransactions" ||
|
|
m == "z_listreceivedbyaddress" || m == "z_listsentbyaddress" || m == "z_getoperationstatus" ||
|
|
m == "z_getoperationresult") {
|
|
c.result = json::array();
|
|
} else if (m == "setban" || m == "clearbanned") {
|
|
c.result = nullptr;
|
|
} else if (m == "z_mergetoaddress") {
|
|
c.errCode = -8; // only the note wallet (FAKE_WALLET_NOTES) can merge; execOne answers it there
|
|
c.errMsg = "fake_dragonxd: set FAKE_WALLET_NOTES to merge notes";
|
|
} else if (m == "z_listaddresses") {
|
|
c.result = json::array({kFakeZAddr});
|
|
} else if (m == "getaddressesbyaccount" || m == "listaddresses") {
|
|
c.result = json::array({kFakeTAddr});
|
|
} else if (m == "getrescaninfo") {
|
|
c.result = {{"rescanning", false}};
|
|
} else if (m == "validateaddress") {
|
|
c.result = {{"isvalid", true}, {"address", param0()}, {"ismine", false}};
|
|
} else if (m == "z_validateaddress") {
|
|
c.result = {{"isvalid", true}, {"address", param0()}, {"type", "sapling"}, {"ismine", false}};
|
|
} else if (m == "z_exportkey" || m == "z_exportmnemonic") {
|
|
c.errCode = -4; // RPC_WALLET_ERROR: keeps seed-derived features (chat) quietly off
|
|
c.errMsg = "fake_dragonxd holds no keys";
|
|
} else {
|
|
c.known = false;
|
|
}
|
|
return c;
|
|
}
|
|
|
|
// dragonxd's warmup allow-list (rpc/server.cpp CRPCTable::execute).
|
|
bool allowedInWarmup(const std::string& m)
|
|
{
|
|
static const char* const kAllowed[] = {
|
|
"stop", "help", "z_listaddresses", "z_exportkey", "getNotarizationsForBlock", "scanNotarizationsDB",
|
|
"getaddressesbyaccount", "listaddresses", "z_exportwallet", "notaries", "signmessage",
|
|
"decoderawtransaction", "dumpprivkey", "getpeerinfo", "getnetworkinfo", "abortrescan"};
|
|
for (const char* a : kAllowed)
|
|
if (m == a) return true;
|
|
return false;
|
|
}
|
|
|
|
std::string helpText(const json& params)
|
|
{
|
|
if (params.empty() || !params[0].is_string())
|
|
return "== Control ==\nhelp ( \"command\" )\nstop\n\n(fake_dragonxd: canned responses only)";
|
|
const std::string cmd = params[0].get<std::string>();
|
|
if (cmd == "stop") return "stop\n\nStop Hush server.";
|
|
if (cmd == "help") return "help ( \"command\" )\n\nList all commands, or get help for a specified command.";
|
|
if (cannedReply(cmd, json::array()).known) return cmd + "\n\n(fake_dragonxd canned method)";
|
|
return "help: unknown command: " + cmd;
|
|
}
|
|
|
|
json rpcReply(const json& result, const json& error, const json& id)
|
|
{
|
|
return {{"result", result}, {"error", error}, {"id", id}};
|
|
}
|
|
|
|
// One JSON-RPC call. `status` gets the HTTP status dragonxd would use (httprpc.cpp JSONErrorReply).
|
|
json execOne(const json& req, int& status)
|
|
{
|
|
const json id = req.is_object() && req.contains("id") ? req["id"] : json(nullptr);
|
|
auto fail = [&](int code, const std::string& msg) {
|
|
status = code == -32600 ? 400 : code == -32601 ? 404 : 500;
|
|
return rpcReply(nullptr, {{"code", code}, {"message", msg}}, id);
|
|
};
|
|
if (!req.is_object() || !req.contains("method") || !req["method"].is_string())
|
|
return fail(-32600, "Missing method");
|
|
const std::string m = req["method"].get<std::string>();
|
|
{
|
|
std::lock_guard<std::mutex> lk(g_methodsMx);
|
|
++g_methodCounts[m];
|
|
}
|
|
json params = req.contains("params") ? req["params"] : json::array();
|
|
if (!params.is_array()) params = json::array();
|
|
status = 200;
|
|
|
|
// Like dragonxd: stop and help take no lock, nor does getrescaninfo (it only reads wallet fields).
|
|
const bool lockFree = (m == "stop" || m == "help" || m == "getrescaninfo");
|
|
if (!lockFree && !cannedReply(m, params).known) return fail(-32601, "Method not found");
|
|
if (g_warmup.load() && !allowedInWarmup(m)) return fail(-28, warmupStatus());
|
|
|
|
if (m == "help") return rpcReply(helpText(params), nullptr, id);
|
|
if (m == "stop") {
|
|
// stop's GenerateBitcoins(false) joins the miner threads, which may be waiting on cs_main.
|
|
if (g_minerOn.load()) {
|
|
std::lock_guard<std::mutex> miners(g_csMain);
|
|
g_minerOn = false;
|
|
}
|
|
requestShutdown(kRpcStop);
|
|
return rpcReply("DRAGONX server stopping...", nullptr, id);
|
|
}
|
|
|
|
Canned c;
|
|
if (lockFree) {
|
|
c = cannedReply(m, params);
|
|
} else {
|
|
std::lock_guard<std::mutex> lk(g_csMain);
|
|
c = cannedReply(m, params);
|
|
if (m == "setgenerate") g_minerOn = !params.empty() && params[0].is_boolean() && params[0].get<bool>();
|
|
if (g_noteWallet) {
|
|
Canned w = noteWalletReply(m, params);
|
|
if (w.known) c = w;
|
|
}
|
|
}
|
|
if (c.errCode != 0) return fail(c.errCode, c.errMsg);
|
|
return rpcReply(c.result, nullptr, id);
|
|
}
|
|
|
|
// =============================================================================================
|
|
// HTTP server: N workers + a bounded queue, like dragonxd's httpserver.cpp WorkQueue
|
|
// =============================================================================================
|
|
|
|
const char* const kTextType = "text/html; charset=ISO-8859-1"; // libevent's default for a bare reply
|
|
std::string g_expectedAuth; // "Basic <base64(user:pass)>"
|
|
std::atomic<bool> g_httpInterrupted{false};
|
|
|
|
std::string base64(const std::string& in)
|
|
{
|
|
static const char* const t = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
|
|
std::string out;
|
|
std::uint32_t val = 0;
|
|
int bits = -6;
|
|
for (unsigned char c : in) {
|
|
val = (val << 8) | c;
|
|
bits += 8;
|
|
while (bits >= 0) {
|
|
out.push_back(t[(val >> bits) & 0x3F]);
|
|
bits -= 6;
|
|
}
|
|
}
|
|
if (bits > -6) out.push_back(t[((val << 8) >> (bits + 8)) & 0x3F]);
|
|
while (out.size() % 4) out.push_back('=');
|
|
return out;
|
|
}
|
|
|
|
struct Item {
|
|
bool isPost = true;
|
|
bool prefill = false;
|
|
std::string body, auth;
|
|
std::mutex mx;
|
|
std::condition_variable cv;
|
|
bool done = false;
|
|
int status = 0;
|
|
std::string contentType, payload;
|
|
};
|
|
using ItemPtr = std::shared_ptr<Item>;
|
|
|
|
void complete(Item& it, int status, const std::string& contentType, const std::string& payload)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lk(it.mx);
|
|
it.status = status;
|
|
it.contentType = contentType;
|
|
it.payload = payload;
|
|
it.done = true;
|
|
}
|
|
it.cv.notify_all();
|
|
if (it.prefill) noteAbandoned(-1);
|
|
}
|
|
|
|
void processItem(Item& it)
|
|
{
|
|
if (!it.isPost) {
|
|
complete(it, 405, kTextType, "JSONRPC server handles only POST requests");
|
|
return;
|
|
}
|
|
if (it.auth != g_expectedAuth) {
|
|
std::this_thread::sleep_for(250ms); // httprpc.cpp: MilliSleep(250) to deter brute forcing
|
|
++g_stats.unauthorized;
|
|
complete(it, 401, "", "");
|
|
return;
|
|
}
|
|
const json req = json::parse(it.body, nullptr, false);
|
|
int status = 200;
|
|
std::string payload;
|
|
if (req.is_discarded()) {
|
|
status = 500;
|
|
payload = rpcReply(nullptr, {{"code", -32700}, {"message", "Parse error"}}, nullptr).dump() + "\n";
|
|
} else if (req.is_array()) {
|
|
json out = json::array();
|
|
for (const auto& r : req) {
|
|
int ignored = 200;
|
|
out.push_back(execOne(r, ignored));
|
|
}
|
|
payload = out.dump() + "\n";
|
|
} else {
|
|
payload = execOne(req, status).dump() + "\n";
|
|
}
|
|
complete(it, status, "application/json", payload);
|
|
}
|
|
|
|
class WorkQueue {
|
|
public:
|
|
void configure(size_t maxDepth, int threads)
|
|
{
|
|
maxDepth_ = maxDepth;
|
|
capacity_ = maxDepth + static_cast<size_t>(threads);
|
|
}
|
|
bool enqueue(const ItemPtr& item)
|
|
{
|
|
std::lock_guard<std::mutex> lk(mx_);
|
|
if (queue_.size() >= maxDepth_) return false;
|
|
queue_.push_back(item);
|
|
raiseMax(g_stats.maxSlots, static_cast<int>(queue_.size()) + inProgress_);
|
|
cv_.notify_one();
|
|
return true;
|
|
}
|
|
size_t slotsInUse()
|
|
{
|
|
std::lock_guard<std::mutex> lk(mx_);
|
|
return queue_.size() + static_cast<size_t>(inProgress_);
|
|
}
|
|
size_t capacity() const { return capacity_; }
|
|
void run()
|
|
{
|
|
for (;;) {
|
|
ItemPtr it;
|
|
{
|
|
std::unique_lock<std::mutex> lk(mx_);
|
|
cv_.wait(lk, [&] { return !running_ || !queue_.empty(); });
|
|
if (!running_) return; // queued items are left for drain(), as in dragonxd
|
|
it = queue_.front();
|
|
queue_.pop_front();
|
|
++inProgress_;
|
|
}
|
|
processItem(*it);
|
|
std::lock_guard<std::mutex> lk(mx_);
|
|
--inProgress_;
|
|
}
|
|
}
|
|
void interrupt()
|
|
{
|
|
std::lock_guard<std::mutex> lk(mx_);
|
|
running_ = false;
|
|
cv_.notify_all();
|
|
}
|
|
std::deque<ItemPtr> drain()
|
|
{
|
|
std::lock_guard<std::mutex> lk(mx_);
|
|
std::deque<ItemPtr> out;
|
|
out.swap(queue_);
|
|
return out;
|
|
}
|
|
|
|
private:
|
|
std::mutex mx_;
|
|
std::condition_variable cv_;
|
|
std::deque<ItemPtr> queue_;
|
|
bool running_ = true;
|
|
int inProgress_ = 0;
|
|
size_t maxDepth_ = 16;
|
|
size_t capacity_ = 24;
|
|
};
|
|
|
|
WorkQueue g_queue;
|
|
|
|
bool waitReadable(socket_t s, int timeoutMs)
|
|
{
|
|
fd_set rf;
|
|
FD_ZERO(&rf);
|
|
FD_SET(s, &rf);
|
|
timeval tv{timeoutMs / 1000, (timeoutMs % 1000) * 1000};
|
|
return select(static_cast<int>(s) + 1, &rf, nullptr, nullptr, &tv) > 0;
|
|
}
|
|
|
|
bool sendAll(socket_t s, const std::string& data)
|
|
{
|
|
size_t off = 0;
|
|
while (off < data.size()) {
|
|
const int n = send(s, data.data() + off, static_cast<int>(std::min<size_t>(data.size() - off, 1 << 20)), MSG_NOSIGNAL);
|
|
if (n <= 0) return false;
|
|
off += static_cast<size_t>(n);
|
|
}
|
|
return true;
|
|
}
|
|
|
|
// The client hung up (e.g. its curl timeout fired) while its request still holds a slot.
|
|
bool peerClosed(socket_t s)
|
|
{
|
|
if (!waitReadable(s, 0)) return false;
|
|
char c;
|
|
return recv(s, &c, 1, MSG_PEEK) <= 0;
|
|
}
|
|
|
|
const char* reasonPhrase(int status)
|
|
{
|
|
switch (status) {
|
|
case 200: return "OK";
|
|
case 400: return "Bad Request";
|
|
case 401: return "Unauthorized";
|
|
case 404: return "Not Found";
|
|
case 405: return "Method Not Allowed";
|
|
case 500: return "Internal Server Error";
|
|
case 503: return "Service Unavailable";
|
|
default: return "Unknown";
|
|
}
|
|
}
|
|
|
|
std::string httpReply(int status, const std::string& contentType, const std::string& body, bool close)
|
|
{
|
|
std::string r = "HTTP/1.1 " + std::to_string(status) + " " + reasonPhrase(status) + "\r\n";
|
|
if (!contentType.empty()) r += "Content-Type: " + contentType + "\r\n";
|
|
if (status == 401) r += "WWW-Authenticate: Basic realm=\"jsonrpc\"\r\n";
|
|
r += "Content-Length: " + std::to_string(body.size()) + "\r\n";
|
|
if (close) r += "Connection: close\r\n";
|
|
r += "\r\n";
|
|
r += body;
|
|
return r;
|
|
}
|
|
|
|
struct HttpRequest {
|
|
std::string method, body, auth;
|
|
bool close = false;
|
|
};
|
|
|
|
enum class ReadResult { Ok, Closed, Timeout, Bad };
|
|
|
|
ReadResult recvMore(socket_t s, std::string& buf, Clock::time_point deadline)
|
|
{
|
|
for (;;) {
|
|
const auto left = std::chrono::duration_cast<std::chrono::milliseconds>(deadline - Clock::now()).count();
|
|
if (left <= 0) return ReadResult::Timeout;
|
|
if (!waitReadable(s, static_cast<int>(std::min<long long>(left, 500)))) continue;
|
|
char tmp[8192];
|
|
const int n = recv(s, tmp, sizeof(tmp), 0);
|
|
if (n <= 0) return ReadResult::Closed;
|
|
buf.append(tmp, static_cast<size_t>(n));
|
|
return ReadResult::Ok;
|
|
}
|
|
}
|
|
|
|
std::string lower(std::string s)
|
|
{
|
|
std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c) { return static_cast<char>(std::tolower(c)); });
|
|
return s;
|
|
}
|
|
|
|
ReadResult readRequest(socket_t s, std::string& buf, HttpRequest& out)
|
|
{
|
|
const auto idle = std::chrono::duration_cast<Clock::duration>(std::chrono::duration<double>(g_cfg.serverTimeoutSecs));
|
|
auto deadline = Clock::now() + idle;
|
|
size_t headerEnd;
|
|
while ((headerEnd = buf.find("\r\n\r\n")) == std::string::npos) {
|
|
if (buf.size() > 64 * 1024) return ReadResult::Bad;
|
|
const ReadResult r = recvMore(s, buf, deadline);
|
|
if (r != ReadResult::Ok) return r;
|
|
}
|
|
const std::string head = buf.substr(0, headerEnd);
|
|
size_t lineEnd = head.find("\r\n");
|
|
const std::string requestLine = head.substr(0, lineEnd);
|
|
out = HttpRequest{};
|
|
out.method = requestLine.substr(0, requestLine.find(' '));
|
|
const bool http10 = requestLine.find("HTTP/1.0") != std::string::npos;
|
|
size_t contentLength = 0;
|
|
bool keepAlive = false;
|
|
while (lineEnd != std::string::npos) {
|
|
const size_t next = head.find("\r\n", lineEnd + 2);
|
|
const std::string line = head.substr(lineEnd + 2, next == std::string::npos ? std::string::npos : next - lineEnd - 2);
|
|
lineEnd = next;
|
|
const size_t colon = line.find(':');
|
|
if (colon == std::string::npos) continue;
|
|
const std::string name = lower(line.substr(0, colon));
|
|
std::string value = line.substr(colon + 1);
|
|
value.erase(0, value.find_first_not_of(" \t"));
|
|
while (!value.empty() && (value.back() == ' ' || value.back() == '\t')) value.pop_back();
|
|
if (name == "content-length") contentLength = static_cast<size_t>(std::strtoul(value.c_str(), nullptr, 10));
|
|
else if (name == "authorization") out.auth = value;
|
|
else if (name == "connection") {
|
|
out.close = lower(value) == "close";
|
|
keepAlive = lower(value) == "keep-alive";
|
|
}
|
|
}
|
|
if (http10 && !keepAlive) out.close = true;
|
|
if (contentLength > 16 * 1024 * 1024) return ReadResult::Bad;
|
|
const size_t total = headerEnd + 4 + contentLength;
|
|
deadline = Clock::now() + idle;
|
|
while (buf.size() < total) {
|
|
const ReadResult r = recvMore(s, buf, deadline);
|
|
if (r != ReadResult::Ok) return r;
|
|
}
|
|
out.body = buf.substr(headerEnd + 4, contentLength);
|
|
buf.erase(0, total);
|
|
return ReadResult::Ok;
|
|
}
|
|
|
|
void serveConnection(socket_t s)
|
|
{
|
|
std::string buf;
|
|
for (;;) {
|
|
HttpRequest req;
|
|
if (readRequest(s, buf, req) != ReadResult::Ok) break;
|
|
++g_stats.requests;
|
|
const bool isStop = req.body.find("\"stop\"") != std::string::npos;
|
|
|
|
// After InterruptHTTPServer, requests on already-open connections get evhttp_send_error(503).
|
|
if (g_httpInterrupted.load()) {
|
|
++g_stats.unavailable;
|
|
if (isStop) event("stop-rejected", "status=503");
|
|
sendAll(s, httpReply(503, "text/html",
|
|
"<HTML><HEAD>\n<TITLE>503 Service Unavailable</TITLE>\n</HEAD><BODY>\n"
|
|
"<H1>Service Unavailable</H1>\n</BODY></HTML>\n",
|
|
true));
|
|
break;
|
|
}
|
|
|
|
auto item = std::make_shared<Item>();
|
|
item->isPost = req.method == "POST";
|
|
item->body = std::move(req.body);
|
|
item->auth = std::move(req.auth);
|
|
if (!g_queue.enqueue(item)) {
|
|
if (++g_stats.queueFull == 1) event("queue-full", "slots=" + std::to_string(g_queue.slotsInUse()));
|
|
if (isStop) event("stop-rejected", "status=500-queue-full");
|
|
if (!sendAll(s, httpReply(500, kTextType, "Work queue depth exceeded", req.close)) || req.close) break;
|
|
continue;
|
|
}
|
|
|
|
bool gone = false;
|
|
for (;;) {
|
|
{
|
|
std::unique_lock<std::mutex> lk(item->mx);
|
|
if (item->cv.wait_for(lk, 200ms, [&] { return item->done; })) break;
|
|
}
|
|
if (!gone && peerClosed(s)) {
|
|
gone = true; // the slot stays taken until a worker reaches it
|
|
noteAbandoned(+1);
|
|
}
|
|
}
|
|
if (gone) {
|
|
noteAbandoned(-1);
|
|
break;
|
|
}
|
|
if (!sendAll(s, httpReply(item->status, item->contentType, item->payload, req.close)) || req.close) break;
|
|
}
|
|
closeSocket(s);
|
|
}
|
|
|
|
void acceptLoop(socket_t listener)
|
|
{
|
|
while (!g_httpInterrupted.load()) {
|
|
if (!waitReadable(listener, 200)) continue;
|
|
sockaddr_in addr{};
|
|
socklen_t len = sizeof(addr);
|
|
const socket_t c = accept(listener, reinterpret_cast<sockaddr*>(&addr), &len);
|
|
if (c == kBadSocket) continue;
|
|
std::thread(serveConnection, c).detach();
|
|
}
|
|
closeSocket(listener); // unbound: new connects are refused (or time out, on Windows)
|
|
}
|
|
|
|
socket_t bindListener(int port)
|
|
{
|
|
const socket_t s = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
|
|
if (s == kBadSocket) return kBadSocket;
|
|
#ifdef _WIN32
|
|
BOOL exclusive = TRUE;
|
|
setsockopt(s, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, reinterpret_cast<const char*>(&exclusive), sizeof(exclusive));
|
|
#else
|
|
int one = 1;
|
|
setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
|
|
#endif
|
|
sockaddr_in a{};
|
|
a.sin_family = AF_INET;
|
|
a.sin_port = htons(static_cast<unsigned short>(port));
|
|
a.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
|
|
if (bind(s, reinterpret_cast<sockaddr*>(&a), sizeof(a)) != 0 || listen(s, 64) != 0) {
|
|
closeSocket(s);
|
|
return kBadSocket;
|
|
}
|
|
return s;
|
|
}
|
|
|
|
// =============================================================================================
|
|
// Background "node" activity
|
|
// =============================================================================================
|
|
|
|
void logUpdateTip(long h)
|
|
{
|
|
const std::time_t blockTime = std::time(nullptr) - 20;
|
|
logf("UpdateTip: new best=%s height=%ld log2_work=56.82813377 tx=%ld date=%s progress=1.000000 cache=1.2MiB(345tx)",
|
|
fakeHash(h).c_str(), h, h * 3 + 12345, utcStamp(blockTime).c_str());
|
|
}
|
|
|
|
// A new block every FAKE_UPDATETIP_SECS; blocks on cs_main like ConnectTip, so none during a rebuild.
|
|
void updateTipLoop()
|
|
{
|
|
while (sleepUnlessShutdown(g_cfg.updateTipSecs)) {
|
|
std::lock_guard<std::mutex> lk(g_csMain);
|
|
if (shutdownRequested()) return; // net threads are interrupted: no more blocks connect
|
|
logUpdateTip(++g_height);
|
|
}
|
|
}
|
|
|
|
// Unrelated lines from other threads, as on the real node (they defeat "newest line" heuristics).
|
|
void noiseLoop()
|
|
{
|
|
static const char* const kLines[] = {
|
|
"socket sending timeout: 1201s",
|
|
"TLS: connection from 203.0.113.10:21768 has been established (initiated)",
|
|
"receive version message: /DragonX:1.3.0/: version 170012, blocks=3327101, us=0.0.0.0:0, peer=17",
|
|
"Using /16 prefix for IP bucketing",
|
|
};
|
|
size_t i = 0;
|
|
while (sleepUnlessShutdown(g_cfg.noiseSecs)) logf("%s", kLines[i++ % (sizeof(kLines) / sizeof(kLines[0]))]);
|
|
}
|
|
|
|
void prefillAbandoned(long n)
|
|
{
|
|
long injected = 0;
|
|
for (; injected < n; ++injected) {
|
|
auto item = std::make_shared<Item>();
|
|
item->prefill = true;
|
|
item->auth = g_expectedAuth;
|
|
item->body = R"({"jsonrpc":"1.0","id":"prefill","method":"getinfo","params":[]})";
|
|
noteAbandoned(+1);
|
|
// Give the workers a moment to take the first items, so the queue depth alone isn't the limit;
|
|
// stop once every slot (workers + queue) is taken.
|
|
bool queued = false;
|
|
while (!(queued = g_queue.enqueue(item)) && g_queue.slotsInUse() < g_queue.capacity())
|
|
std::this_thread::sleep_for(5ms);
|
|
if (!queued) {
|
|
noteAbandoned(-1);
|
|
break;
|
|
}
|
|
}
|
|
event("prefill", "n=" + std::to_string(injected));
|
|
}
|
|
|
|
// BuildWitnessCache: Phase 1 logs "Reading blocks for witness rebuild: <offset> / <total>" whenever
|
|
// the ABSOLUTE height is a multiple of 5000, and aborts on shutdown. Phase 2 logs nothing and ignores
|
|
// shutdown. Both hold cs_main.
|
|
void rebuildLoop()
|
|
{
|
|
if (!sleepUnlessShutdown(g_cfg.rebuildDelaySecs)) return;
|
|
std::unique_lock<std::mutex> lk(g_csMain);
|
|
if (shutdownRequested()) return;
|
|
const auto started = Clock::now();
|
|
const long endHeight = g_height.load();
|
|
const long startHeight = endHeight - g_cfg.rebuildTotal;
|
|
event("rebuild-start", "total=" + std::to_string(g_cfg.rebuildTotal) + " secs=" + std::to_string(g_cfg.rebuildSecs));
|
|
if (g_cfg.prefill > 0) std::thread(prefillAbandoned, g_cfg.prefill).detach();
|
|
|
|
std::vector<long> offsets;
|
|
for (long h = (startHeight / 5000 + 1) * 5000; h <= endHeight; h += 5000) offsets.push_back(h - startHeight);
|
|
const double step = offsets.empty() ? 0 : g_cfg.phase1Secs / static_cast<double>(offsets.size());
|
|
for (long off : offsets) {
|
|
if (shutdownRequested()) {
|
|
logf("BuildWitnessCache: shutdown requested, aborting witness rebuild");
|
|
event("rebuild-aborted");
|
|
return;
|
|
}
|
|
if (!g_cfg.rebuildSilent) logf("Reading blocks for witness rebuild: %ld / %ld", off, g_cfg.rebuildTotal);
|
|
std::this_thread::sleep_for(std::chrono::duration<double>(step));
|
|
}
|
|
|
|
const auto phase2End = started + std::chrono::duration_cast<Clock::duration>(std::chrono::duration<double>(g_cfg.rebuildSecs));
|
|
std::this_thread::sleep_until(phase2End); // Phase 2: silent, and deaf to shutdown
|
|
const long long ms = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - started).count();
|
|
if (!g_cfg.rebuildSilent)
|
|
logf("BuildWitnessCache: rebuilt %ld note witness cache(s) to height %ld in %lldms using %d thread(s)",
|
|
g_cfg.notes, endHeight, ms, 8);
|
|
event("rebuild-end", "ms=" + std::to_string(ms));
|
|
if (g_cfg.postRebuildLockSecs > 0)
|
|
std::this_thread::sleep_for(std::chrono::duration<double>(g_cfg.postRebuildLockSecs));
|
|
}
|
|
|
|
// =============================================================================================
|
|
// Datadir lock, pid file, cookie
|
|
// =============================================================================================
|
|
|
|
#ifdef _WIN32
|
|
HANDLE g_lockHandle = INVALID_HANDLE_VALUE;
|
|
#else
|
|
int g_lockFd = -1;
|
|
#endif
|
|
|
|
bool lockDatadir(const fs::path& datadir)
|
|
{
|
|
const std::string path = (datadir / ".lock").string();
|
|
#ifdef _WIN32
|
|
g_lockHandle = CreateFileA(path.c_str(), GENERIC_READ | GENERIC_WRITE,
|
|
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, nullptr, OPEN_ALWAYS,
|
|
FILE_ATTRIBUTE_NORMAL, nullptr);
|
|
if (g_lockHandle == INVALID_HANDLE_VALUE) return false;
|
|
OVERLAPPED ov{};
|
|
return LockFileEx(g_lockHandle, LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY, 0, MAXDWORD, MAXDWORD, &ov) != 0;
|
|
#else
|
|
g_lockFd = ::open(path.c_str(), O_RDWR | O_CREAT, 0644);
|
|
if (g_lockFd < 0) return false;
|
|
struct flock fl{};
|
|
fl.l_type = F_WRLCK;
|
|
fl.l_whence = SEEK_SET;
|
|
return fcntl(g_lockFd, F_SETLK, &fl) == 0; // what boost::interprocess::file_lock uses
|
|
#endif
|
|
}
|
|
|
|
std::string randomHex(size_t bytes)
|
|
{
|
|
std::random_device rd;
|
|
std::string out;
|
|
char b[3];
|
|
for (size_t i = 0; i < bytes; ++i) {
|
|
std::snprintf(b, sizeof(b), "%02x", rd() & 0xFF);
|
|
out += b;
|
|
}
|
|
return out;
|
|
}
|
|
|
|
[[noreturn]] void exitNow(int code)
|
|
{
|
|
g_log.close();
|
|
{
|
|
std::lock_guard<std::mutex> lk(g_eventsMx);
|
|
if (g_events) std::fclose(g_events);
|
|
g_events = nullptr;
|
|
}
|
|
std::fflush(stdout);
|
|
std::_Exit(code);
|
|
}
|
|
|
|
[[noreturn]] void initError(const std::string& msg)
|
|
{
|
|
logf("Error: %s", msg.c_str());
|
|
std::fprintf(stderr, "Error: %s\n", msg.c_str());
|
|
event("init-error", "msg=\"" + msg + "\"");
|
|
exitNow(1);
|
|
}
|
|
|
|
void setThreadName(const char* name)
|
|
{
|
|
#ifdef __linux__
|
|
prctl(PR_SET_NAME, name, 0, 0, 0); // the main thread's name is /proc/<pid>/comm
|
|
#else
|
|
(void)name;
|
|
#endif
|
|
}
|
|
|
|
void loadConfig(int argc, char** argv)
|
|
{
|
|
const auto args = parseArgs(argc, argv);
|
|
g_cfg.datadir = args.count("datadir") ? fs::path(args.at("datadir")) : defaultDatadir();
|
|
std::error_code ec;
|
|
fs::create_directories(g_cfg.datadir, ec);
|
|
const auto conf = parseConf(g_cfg.datadir / "DRAGONX.conf");
|
|
auto get = [&](const std::string& key, const std::string& def) {
|
|
if (args.count(key)) return args.at(key);
|
|
if (conf.count(key)) return conf.at(key);
|
|
return def;
|
|
};
|
|
g_cfg.printToConsole = get("printtoconsole", "0") != "0";
|
|
g_cfg.rpcPort = std::atoi(get("rpcport", "21769").c_str());
|
|
g_cfg.rpcThreads = std::max(1, std::atoi(get("rpcthreads", "8").c_str()));
|
|
g_cfg.rpcWorkQueue = std::max(1, std::atoi(get("rpcworkqueue", "16").c_str()));
|
|
g_cfg.rpcUser = get("rpcuser", "");
|
|
g_cfg.rpcPassword = get("rpcpassword", "");
|
|
|
|
g_cfg.warmupSecs = envDouble("FAKE_WARMUP_SECS", g_cfg.warmupSecs);
|
|
g_cfg.replaySecs = envDouble("FAKE_REPLAY_SECS", g_cfg.replaySecs);
|
|
g_cfg.replayBlocksPerSec = std::max(0.1, envDouble("FAKE_REPLAY_BPS", g_cfg.replayBlocksPerSec));
|
|
g_cfg.rebuildDelaySecs = envDouble("FAKE_REBUILD_DELAY_SECS", g_cfg.rebuildDelaySecs);
|
|
g_cfg.rebuildSecs = envDouble("FAKE_REBUILD_SECS", g_cfg.rebuildSecs);
|
|
g_cfg.phase1Secs = std::min(envDouble("FAKE_PHASE1_SECS", g_cfg.phase1Secs), g_cfg.rebuildSecs);
|
|
g_cfg.postRebuildLockSecs = envDouble("FAKE_POST_REBUILD_LOCK_SECS", g_cfg.postRebuildLockSecs);
|
|
g_cfg.flushSecs = envDouble("FAKE_FLUSH_SECS", g_cfg.flushSecs);
|
|
g_cfg.updateTipSecs = std::max(1.0, envDouble("FAKE_UPDATETIP_SECS", g_cfg.updateTipSecs));
|
|
g_cfg.noiseSecs = std::max(1.0, envDouble("FAKE_NOISE_SECS", g_cfg.noiseSecs));
|
|
g_cfg.serverTimeoutSecs = std::max(1.0, envDouble("FAKE_SERVER_TIMEOUT_SECS", g_cfg.serverTimeoutSecs));
|
|
g_cfg.prefill = envLong("FAKE_PREFILL", g_cfg.prefill);
|
|
g_cfg.height = envLong("FAKE_HEIGHT", g_cfg.height);
|
|
g_cfg.rebuildTotal = std::max(1L, envLong("FAKE_REBUILD_TOTAL", g_cfg.rebuildTotal));
|
|
g_cfg.notes = envLong("FAKE_NOTES", g_cfg.notes);
|
|
g_cfg.version = envLong("FAKE_VERSION", g_cfg.version);
|
|
g_cfg.stuckShutdown = envLong("FAKE_STUCK_SHUTDOWN", 0) != 0;
|
|
g_cfg.mining = envLong("FAKE_MINING", 0) != 0;
|
|
g_cfg.rebuildSilent = envLong("FAKE_REBUILD_SILENT", 0) != 0;
|
|
}
|
|
|
|
std::string knobsLine()
|
|
{
|
|
char buf[512];
|
|
std::snprintf(buf, sizeof(buf),
|
|
"warmup=%gs replay=%gs rebuild=%gs after=%gs phase1=%gs post_lock=%gs flush=%gs tip=%gs "
|
|
"prefill=%ld version=%ld workers=%d depth=%d port=%d console=%d mining=%d silent=%d",
|
|
g_cfg.warmupSecs, g_cfg.replaySecs, g_cfg.rebuildSecs, g_cfg.rebuildDelaySecs, g_cfg.phase1Secs,
|
|
g_cfg.postRebuildLockSecs, g_cfg.flushSecs, g_cfg.updateTipSecs, g_cfg.prefill, g_cfg.version,
|
|
g_cfg.rpcThreads, g_cfg.rpcWorkQueue, g_cfg.rpcPort, g_cfg.printToConsole ? 1 : 0,
|
|
g_cfg.mining ? 1 : 0, g_cfg.rebuildSilent ? 1 : 0);
|
|
return buf;
|
|
}
|
|
|
|
} // namespace
|
|
|
|
int main(int argc, char** argv)
|
|
{
|
|
#ifdef _WIN32
|
|
WSADATA wsa;
|
|
WSAStartup(MAKEWORD(2, 2), &wsa);
|
|
#else
|
|
std::signal(SIGPIPE, SIG_IGN); // dragonxd ignores SIGPIPE too (init.cpp)
|
|
std::signal(SIGHUP, SIG_IGN);
|
|
#endif
|
|
setThreadName("dragonxd");
|
|
loadConfig(argc, argv);
|
|
g_log.open(g_cfg.datadir, g_cfg.printToConsole);
|
|
g_events = std::fopen((g_cfg.datadir / "fake_dragonxd.events").string().c_str(), "ab");
|
|
#ifdef _WIN32
|
|
const long pid = static_cast<long>(GetCurrentProcessId());
|
|
#else
|
|
const long pid = static_cast<long>(getpid());
|
|
#endif
|
|
|
|
const std::string ver = "v" + std::to_string(g_cfg.version / 1000000) + "." +
|
|
std::to_string(g_cfg.version / 10000 % 100) + "." +
|
|
std::to_string(g_cfg.version / 100 % 100);
|
|
logf("DragonX version %s-fakestub", ver.c_str());
|
|
logf("fake_dragonxd: %s", knobsLine().c_str());
|
|
logf("Using data directory %s", g_cfg.datadir.string().c_str());
|
|
event("start", "pid=" + std::to_string(pid) + " " + knobsLine());
|
|
|
|
if (!lockDatadir(g_cfg.datadir))
|
|
initError("Cannot obtain a lock on data directory " + g_cfg.datadir.string() +
|
|
". Full node is probably already running.");
|
|
|
|
if (g_cfg.stuckShutdown) {
|
|
// The Sep 29 orphan: an old node wedged in shutdown. It holds the datadir lock, has already
|
|
// unbound RPC, and nothing short of a hard kill ends it (on POSIX SIGTERM just re-sets
|
|
// fRequestShutdown, which changes nothing).
|
|
#ifndef _WIN32
|
|
std::signal(SIGTERM, SIG_IGN);
|
|
std::signal(SIGINT, SIG_IGN);
|
|
#endif
|
|
logf("fake_dragonxd: FAKE_STUCK_SHUTDOWN=1, holding the datadir lock with no RPC listener until killed");
|
|
event("stuck");
|
|
for (;;) std::this_thread::sleep_for(1h);
|
|
}
|
|
|
|
#ifndef _WIN32
|
|
std::signal(SIGTERM, onTermSignal);
|
|
std::signal(SIGINT, onTermSignal);
|
|
const fs::path pidFile = g_cfg.datadir / "hushd.pid";
|
|
{
|
|
std::ofstream(pidFile) << pid << "\n";
|
|
}
|
|
#endif
|
|
|
|
fs::path cookieFile;
|
|
if (g_cfg.rpcPassword.empty()) {
|
|
g_cfg.rpcUser = "__cookie__";
|
|
g_cfg.rpcPassword = randomHex(32);
|
|
cookieFile = g_cfg.datadir / ".cookie";
|
|
std::ofstream(cookieFile) << g_cfg.rpcUser << ":" << g_cfg.rpcPassword;
|
|
logf("Generated RPC authentication cookie %s", cookieFile.string().c_str());
|
|
}
|
|
g_expectedAuth = "Basic " + base64(g_cfg.rpcUser + ":" + g_cfg.rpcPassword);
|
|
g_height = g_cfg.height;
|
|
loadNoteWallet(); // FAKE_WALLET_NOTES (optional)
|
|
g_minerOn = g_cfg.mining;
|
|
|
|
const socket_t listener = bindListener(g_cfg.rpcPort);
|
|
if (listener == kBadSocket) initError("Unable to bind any endpoint for RPC server");
|
|
g_queue.configure(static_cast<size_t>(g_cfg.rpcWorkQueue), g_cfg.rpcThreads);
|
|
logf("HTTP: creating work queue of depth %d", g_cfg.rpcWorkQueue);
|
|
logf("HTTP: starting %d worker threads", g_cfg.rpcThreads);
|
|
std::vector<std::thread> workers;
|
|
for (int i = 0; i < g_cfg.rpcThreads; ++i) workers.emplace_back([] { g_queue.run(); });
|
|
std::thread acceptThread(acceptLoop, listener);
|
|
event("listening", "port=" + std::to_string(g_cfg.rpcPort));
|
|
|
|
// ---- Init: warmup, optional startup replay, then "Done Loading" -----------------------------
|
|
std::thread tipThread, noiseThread, rebuildThread;
|
|
setInitMessage("Loading block index...");
|
|
bool initOk = sleepUnlessShutdown(g_cfg.warmupSecs);
|
|
if (initOk && g_cfg.replaySecs > 0) {
|
|
// ActivateBestChain: cs_main per step, checks ShutdownRequested between steps.
|
|
setInitMessage("Activating best chain...");
|
|
event("replay-start");
|
|
const auto end = Clock::now() + std::chrono::duration_cast<Clock::duration>(std::chrono::duration<double>(g_cfg.replaySecs));
|
|
const auto stepDur = std::chrono::duration<double>(1.0 / g_cfg.replayBlocksPerSec);
|
|
while (initOk && Clock::now() < end) {
|
|
{
|
|
std::lock_guard<std::mutex> lk(g_csMain);
|
|
logUpdateTip(++g_height);
|
|
}
|
|
initOk = sleepUnlessShutdown(stepDur.count());
|
|
}
|
|
event(initOk ? "replay-end" : "replay-interrupted");
|
|
}
|
|
if (initOk) {
|
|
setInitMessage("Full Node Done Loading! :)");
|
|
g_warmup = false;
|
|
event("warmup-end");
|
|
tipThread = std::thread(updateTipLoop);
|
|
noiseThread = std::thread(noiseLoop);
|
|
if (g_cfg.rebuildSecs > 0) rebuildThread = std::thread(rebuildLoop);
|
|
}
|
|
|
|
// ---- Run until `stop` / SIGTERM --------------------------------------------------------------
|
|
auto lastStats = Clock::now();
|
|
std::string lastStatsLine;
|
|
while (!shutdownRequested()) {
|
|
std::this_thread::sleep_for(50ms);
|
|
if (Clock::now() - lastStats >= 30s) {
|
|
lastStats = Clock::now();
|
|
const std::string s = statsLine();
|
|
if (s != lastStatsLine) {
|
|
event("stats", s);
|
|
event("methods", topMethodsLine(10));
|
|
}
|
|
lastStatsLine = s;
|
|
}
|
|
}
|
|
|
|
// ---- Shutdown ----------------------------------------------------------------------------------
|
|
// Interrupt(): unbind RPC and reject new requests on open connections; workers finish only the
|
|
// item they hold. Then the thread-group join waits for the rebuild (Phase 2 can't be aborted),
|
|
// and only then does Shutdown() log "In progress...".
|
|
g_httpInterrupted = true;
|
|
g_queue.interrupt();
|
|
acceptThread.join();
|
|
event("http-interrupted");
|
|
logf("tor: Thread interrupt");
|
|
logf("scheduler thread interrupt");
|
|
logf("net thread interrupt");
|
|
if (noiseThread.joinable()) noiseThread.join();
|
|
if (tipThread.joinable()) tipThread.join();
|
|
if (rebuildThread.joinable()) rebuildThread.join();
|
|
|
|
logf("Shutdown: In progress...");
|
|
setThreadName("hush-shutoff");
|
|
event("shutdown-begin");
|
|
for (auto& w : workers) w.join();
|
|
for (const auto& it : g_queue.drain()) { // StopHTTPServer: ~HTTPRequest on never-run items
|
|
logf("~HTTPRequest: Unhandled request");
|
|
++g_stats.unhandled;
|
|
complete(*it, 500, kTextType, "Unhandled request");
|
|
}
|
|
std::this_thread::sleep_for(std::chrono::duration<double>(g_cfg.flushSecs));
|
|
logf("Shutdown: done");
|
|
#ifndef _WIN32
|
|
std::error_code ec;
|
|
fs::remove(pidFile, ec);
|
|
#endif
|
|
if (!cookieFile.empty()) {
|
|
std::error_code ec2;
|
|
fs::remove(cookieFile, ec2);
|
|
}
|
|
std::this_thread::sleep_for(200ms); // let connection threads flush their last replies
|
|
event("methods", topMethodsLine(10));
|
|
event("clean-exit", statsLine());
|
|
exitNow(0);
|
|
}
|