From 541223daa6e6510daf3bdfdcb4dd4f6b0e38d877 Mon Sep 17 00:00:00 2001 From: bitflash-maintainer Date: Sat, 8 Aug 2026 02:29:26 +0000 Subject: pool: sync-gate job creation, retry failed payouts; widen nostr relay set - RebuildJob: no jobs with zero peers or a tip older than 30min; mining unsynced produced orphaned blocks and payouts that can never be paid (observed in production: 8 orphan 'found blocks') - payout exec: on SendMoney failure requeue with 60s backoff, up to 30 attempts, instead of silently forfeiting the payment - nostr: add oxtr.dev, nos.social, fmt.wiz.biz, current.fyi, nostr.bg (12 relays total) for announcement redundancy (cherry picked from commit f1986cd7e4a6f69469a2256eb96ff03f3f146792) --- src/btfrv.cpp | 41 ++++--- src/main.cpp | 318 +++++++++++++++++++++++++++++++++++++++++++++++-------- src/main.h | 3 + src/main_gui.cpp | 40 +++++++ src/net.cpp | 36 ++++++- src/nostr.cpp | 5 + src/rpc.cpp | 95 ++++++++++++++++- 7 files changed, 471 insertions(+), 67 deletions(-) (limited to 'src') diff --git a/src/btfrv.cpp b/src/btfrv.cpp index 6ac3280..905717f 100644 --- a/src/btfrv.cpp +++ b/src/btfrv.cpp @@ -94,31 +94,42 @@ static SOCKET ConnectTo(const char* host, unsigned short port) { struct addrinfo hints, *res = NULL; memset(&hints, 0, sizeof(hints)); - hints.ai_family = AF_INET; + hints.ai_family = AF_UNSPEC; // IPv4 or IPv6 (some relays are v6-only) hints.ai_socktype = SOCK_STREAM; char portstr[16]; sprintf(portstr, "%u", (unsigned)port); if (getaddrinfo(host, portstr, &hints, &res) != 0 || !res) return INVALID_SOCKET; - SOCKET s = BtfSocketTag(socket(res->ai_family, res->ai_socktype, res->ai_protocol), SOCK_RV_DIAL); - if (s == INVALID_SOCKET) { freeaddrinfo(res); return INVALID_SOCKET; } - // Apply a short timeout only for the connect() and initial handshake so an - // unresponsive relay doesn't block the thread. Once the tunnel is live the - // timeout is cleared -- a forwarding pipe must block indefinitely on recv(). + // Try each resolved address until one connects. + SOCKET s = INVALID_SOCKET; + for (struct addrinfo* rp = res; rp != NULL; rp = rp->ai_next) + { + s = BtfSocketTag(socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol), SOCK_RV_DIAL); + if (s == INVALID_SOCKET) + continue; + + // Apply a short timeout only for the connect() and initial handshake so an + // unresponsive relay doesn't block the thread. Once the tunnel is live the + // timeout is cleared -- a forwarding pipe must block indefinitely on recv(). #ifdef _WIN32 - DWORD tvShort = 10000; - setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tvShort, sizeof(tvShort)); - setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, (const char*)&tvShort, sizeof(tvShort)); + DWORD tvShort = 10000; + setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tvShort, sizeof(tvShort)); + setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, (const char*)&tvShort, sizeof(tvShort)); #else - struct timeval tvShort; tvShort.tv_sec = 10; tvShort.tv_usec = 0; - setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, &tvShort, sizeof(tvShort)); - setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, &tvShort, sizeof(tvShort)); + struct timeval tvShort; tvShort.tv_sec = 10; tvShort.tv_usec = 0; + setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, (char*)&tvShort, sizeof(tvShort)); + setsockopt(s, SOL_SOCKET, SO_SNDTIMEO, (char*)&tvShort, sizeof(tvShort)); #endif - if (connect(s, res->ai_addr, (int)res->ai_addrlen) != 0) - { - CLOSESOCK(s); freeaddrinfo(res); return INVALID_SOCKET; + if (connect(s, rp->ai_addr, (int)rp->ai_addrlen) != 0) + { + CLOSESOCK(s); + s = INVALID_SOCKET; + continue; + } + break; } freeaddrinfo(res); + if (s == INVALID_SOCKET) return INVALID_SOCKET; // Clear the timeout -- the socket is now a live tunnel and must not // time out when idle. recv() will block until data arrives or peer closes. diff --git a/src/main.cpp b/src/main.cpp index 0342650..32ab952 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -102,6 +102,9 @@ CAddress addrIncoming; int nMineMode = MINE_RELAY; bool fMineModeFromCommandLine = false; string strParticipantPool; // participant mode: pool .btf address +string strPayoutAddress; // participant: /payoutaddr override (mine without a wallet) +string strPayoutFile; // participant: /payoutfile with one address per line, rotated per reconnect +int nHashThreads = 0; // participant: /hashthreads override (0 = cores-1) string strPoolName = "Bitflash Pool"; string strPoolDashboardUrl; double dPoolFeePercent = 0.0; @@ -3488,6 +3491,102 @@ static bool StratumRecvLine(SOCKET s, std::string& buf, // How long to wait for a mining.notify after authorize before giving up static const int64 STUCK_NO_JOB_TIMEOUT_SECS = 120; +// ------------------------------------------------------------------ +// Multi-threaded pool worker. The original loop hashed a single nonce +// per blocking 50ms socket poll, capping one instance at ~20 H/s no +// matter how many cores the machine has. The socket and job bookkeeping +// stay on the calling thread; N hash threads grind nonce ranges on their +// own RandomX VMs (all sharing the one ~2 GB dataset) and drop found +// shares on a queue the I/O thread drains and submits. +// ------------------------------------------------------------------ +struct ParticipantJobShared +{ + std::mutex mtx; + unsigned char header[80]; + uint256 target; + std::string jobId; + uint64 jobSeq; + bool haveJob; + bool stop; + ParticipantJobShared() : jobSeq(0), haveJob(false), stop(false) + { + memset(header, 0, sizeof(header)); + } +}; + +struct ParticipantShareHit +{ + uint64 jobSeq; + std::string jobId; + unsigned int nonce; +}; + +static void ParticipantHashWorker(ParticipantJobShared* sh, int idx, int nWorkers, + std::vector* hits, + std::mutex* hitsMtx) +{ + void* vm = RandomXCreateMinerVM(); + if (!vm) + { + LogPrint("worker", "[worker] hash thread %d: RandomX VM creation failed\n", idx); + return; + } + + unsigned char header[80]; + uint256 target; + std::string jobId; + uint64 mySeq = 0; + bool active = false; + unsigned int nNonce = 0; + uint64 localHashes = 0; + + while (!fShutdown) + { + { + std::lock_guard lk(sh->mtx); + if (sh->stop) + break; + if (sh->haveJob && sh->jobSeq != mySeq) + { + memcpy(header, sh->header, 80); + target = sh->target; + jobId = sh->jobId; + mySeq = sh->jobSeq; + nNonce = (unsigned int)(idx + 1); + active = true; + } + } + if (!active) + { + Sleep(50); + continue; + } + + memcpy(header + 76, &nNonce, 4); + uint256 hash = RandomXHashWithVM(vm, header, 80); + localHashes++; + if ((localHashes & 0x3ff) == 0) + gParticipantHashes.fetch_add(0x400); + + if (hash <= target) + { + LogPrint("worker", "[worker] share found! nonce=%u job=%s (thread %d)\n", + nNonce, jobId.c_str(), idx); + std::lock_guard lk(*hitsMtx); + hits->push_back(ParticipantShareHit{mySeq, jobId, nNonce}); + } + + unsigned int next = nNonce + (unsigned int)nWorkers; + if (next < nNonce) // wrapped past 2^32: range exhausted, wait for next job + active = false; + else + nNonce = next; + } + + gParticipantHashes.fetch_add(localHashes & 0x3ff); + RandomXDestroyMinerVM(vm); +} + static bool PoolParticipantMiner() { // ------------------------------------------------------------------ @@ -3502,6 +3601,51 @@ static bool PoolParticipantMiner() LogPrint("worker", "[worker] connecting to pool %s\n", strParticipantPool.c_str()); SetParticipantMiningStatus("resolving pool address"); + // ------------------------------------------------------------------ + // Steps 1-2: connect to the pool. Two address forms: + // tcp:HOST:PORT -- plain TCP straight to the pool's stratum port + // (e.g. our own operator node on this machine), + // no .btf resolution or rendezvous tunnel involved + // .btf -- resolve via Nostr, dial through a rendezvous relay + // ------------------------------------------------------------------ + btf_socket_t s = INVALID_SOCKET; + if (strParticipantPool.compare(0, 4, "tcp:") == 0) + { + std::string hp = strParticipantPool.substr(4); + size_t colon = hp.rfind(':'); + if (colon == std::string::npos) { + LogPrint("worker", "[worker] bad tcp pool address '%s' (want tcp:host:port)\n", + strParticipantPool.c_str()); + SetParticipantMiningStatus("bad pool address"); + return false; + } + std::string host = hp.substr(0, colon); + int port = atoi(hp.substr(colon + 1).c_str()); + unsigned long ip = inet_addr(host.c_str()); + if (port <= 0 || port > 65535 || ip == INADDR_NONE) { + LogPrint("worker", "[worker] bad tcp pool endpoint '%s'\n", hp.c_str()); + SetParticipantMiningStatus("bad pool endpoint"); + return false; + } + SetParticipantMiningStatus("connecting to pool (direct tcp)"); + struct sockaddr_in sa; + memset(&sa, 0, sizeof(sa)); + sa.sin_family = AF_INET; + sa.sin_port = htons((unsigned short)port); + sa.sin_addr.s_addr = ip; + SOCKET cs = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (cs == INVALID_SOCKET || connect(cs, (struct sockaddr*)&sa, sizeof(sa)) == SOCKET_ERROR) { + LogPrint("worker", "[worker] direct connect to %s FAILED\n", hp.c_str()); + SetParticipantMiningStatus("pool connect failed"); + if (cs != INVALID_SOCKET) closesocket(cs); + return false; + } + LogPrint("worker", "[worker] connected directly to pool %s\n", hp.c_str()); + s = (btf_socket_t)cs; + goto connected; + } + + { unsigned char target_pubkey[32]; if (!btf::ParseAddress(strParticipantPool, target_pubkey)) { LogPrint("worker", "[worker] invalid .btf address: %s\n", strParticipantPool.c_str()); @@ -3536,11 +3680,11 @@ static bool PoolParticipantMiner() } // ------------------------------------------------------------------ - // Step 2: open encrypted tunnel to pool + // Step 2b: open encrypted tunnel to pool // ------------------------------------------------------------------ SetParticipantMiningStatus("opening tunnel"); - btf_socket_t s = btf::BtfClientTunnel(host.c_str(), (unsigned short)port, - target_pubkey, service_enc_pub); + s = btf::BtfClientTunnel(host.c_str(), (unsigned short)port, + target_pubkey, service_enc_pub); if (s == INVALID_SOCKET) { LogPrint("worker", "[worker] tunnel connect FAILED via %s" " (relay unreachable or pool not listening)\n", @@ -3550,11 +3694,48 @@ static bool PoolParticipantMiner() } LogPrint("worker", "[worker] tunnel open via %s\n", meetingHostPort.c_str()); SetParticipantMiningStatus("tunnel open"); + } + +connected: // ------------------------------------------------------------------ - // Step 3: determine our payout address (mining username) + // Step 3: determine our payout address (mining username). /payoutaddr + // wins -- it lets a machine mine to an address whose keys live + // elsewhere, so no wallet file is needed on the mining machine. + // /payoutfile rotates through a list of such addresses: one is picked + // per (re)connection, and the session self-limits (see rotation break + // in the main loop) so consecutive rounds land on fresh addresses. // ------------------------------------------------------------------ - std::string username; + static std::vector vPayoutAddrs; + static size_t nPayoutRot = 0; + if (!strPayoutFile.empty() && vPayoutAddrs.empty()) + { + FILE* f = fopen(strPayoutFile.c_str(), "r"); + if (f) + { + char line[256]; + while (fgets(line, sizeof(line), f)) + { + std::string a = line; + while (!a.empty() && (a.back() == '\n' || a.back() == '\r' || + a.back() == ' ' || a.back() == '\t')) + a.pop_back(); + if (!a.empty()) + vPayoutAddrs.push_back(a); + } + fclose(f); + LogPrint("worker", "[worker] loaded %zu payout addresses from %s\n", + vPayoutAddrs.size(), strPayoutFile.c_str()); + } + else + LogPrint("worker", "[worker] cannot open payout file %s\n", + strPayoutFile.c_str()); + } + + std::string username = strPayoutAddress; + if (username.empty() && !vPayoutAddrs.empty()) + username = vPayoutAddrs[nPayoutRot++ % vPayoutAddrs.size()]; + if (username.empty()) { std::vector vchPubKey; if (CWalletDB("r").ReadDefaultKey(vchPubKey)) @@ -3595,13 +3776,7 @@ static bool PoolParticipantMiner() unsigned int nThreads = std::thread::hardware_concurrency(); RandomXInitDataset(nThreads > 0 ? (int)nThreads : 1); } - void* rxvm = RandomXCreateMinerVM(); - if (!rxvm) { - LogPrint("worker", "[worker] failed to create RandomX VM\n"); - SetParticipantMiningStatus("RandomX init failed"); - closesocket(s); return false; - } - LogPrint("worker", "[worker] RandomX VM ready (%s)\n", + LogPrint("worker", "[worker] RandomX dataset ready (%s)\n", RandomXFastReady() ? "fast 2GB" : "light 256MB"); // ------------------------------------------------------------------ @@ -3619,7 +3794,6 @@ static bool PoolParticipantMiner() {"params",json::array()}})) { LogPrint("worker", "[worker] subscribe send FAILED -- pool dropped connection\n"); SetParticipantMiningStatus("subscribe send failed"); - RandomXDestroyMinerVM(rxvm); closesocket(s); return false; } @@ -3629,10 +3803,29 @@ static bool PoolParticipantMiner() {"params",json::array({username,"x"})}})) { LogPrint("worker", "[worker] authorize send FAILED -- pool dropped connection\n"); SetParticipantMiningStatus("authorize send failed"); - RandomXDestroyMinerVM(rxvm); closesocket(s); return false; } + // ------------------------------------------------------------------ + // Step 6: spawn hash threads (each with its own RandomX VM on the + // shared dataset), then run the I/O loop below: receive pool messages, + // publish jobs to the threads, submit the shares they find. + // ------------------------------------------------------------------ + int nWorkers = nHashThreads > 0 ? nHashThreads + : (int)std::thread::hardware_concurrency(); + if (nWorkers < 2) nWorkers = 2; + if (nHashThreads <= 0) nWorkers -= 1; // leave one core for node + I/O threads + ParticipantJobShared jobShared; + std::vector shareHits; + std::mutex shareHitsMtx; + std::vector hashThreads; + for (int i = 0; i < nWorkers; i++) + hashThreads.emplace_back(ParticipantHashWorker, &jobShared, i, nWorkers, + &shareHits, &shareHitsMtx); + LogPrint("worker", "[worker] spawned %d hash threads on shared RandomX dataset\n", + nWorkers); + const uint64 sessionStartHashes = gParticipantHashes.load(); + // ------------------------------------------------------------------ // Step 6: main loop -- receive messages, hash, submit shares // ------------------------------------------------------------------ @@ -3650,6 +3843,8 @@ static bool PoolParticipantMiner() uint64 sessionSharesSent = 0; // this connection only uint64 sessionSharesAccepted = 0; int64 sampleStart = GetTime(); + int64 connStart = GetTime(); + const int64 PAYOUT_ROTATE_SECS = 300; // reconnect with next /payoutfile address SetParticipantMiningStatus("waiting for pool response"); @@ -3757,6 +3952,14 @@ static bool PoolParticipantMiner() memcpy(&target, tgtBytes.data(), 32); haveJob = true; nNonce = 1; + { + std::lock_guard lk(jobShared.mtx); + memcpy(jobShared.header, header, 80); + jobShared.target = target; + jobShared.jobId = jobId; + jobShared.jobSeq++; + jobShared.haveJob = true; + } LogPrint("worker", "[pool->worker] new job %s received\n", jobId.c_str()); SetParticipantMiningStatus("hashing job " + jobId); @@ -3812,46 +4015,71 @@ static bool PoolParticipantMiner() } } + // -- Payout address rotation: end this session cleanly so the outer + // loop reconnects and authorizes the next /payoutfile address -- + if (!vPayoutAddrs.empty() && GetTime() - connStart > PAYOUT_ROTATE_SECS) + { + LogPrint("worker", "[worker] payout rotation interval reached" + " -- reconnecting with next address\n"); + break; + } + if (!haveJob) { Sleep(100); continue; } - // -- Hash one nonce -- - memcpy(header+76, &nNonce, 4); - uint256 hash = RandomXHashWithVM(rxvm, header, 80); - hashesSinceSample++; - gParticipantHashes.fetch_add(1); - if ((hashesSinceSample & 0x3ff) == 0) - UpdateParticipantHashRate(hashesSinceSample, sampleStart); - - if (hash <= target) { - LogPrint("worker", "[worker] share found! nonce=%u job=%s difficulty=%.4f\n", - nNonce, jobId.c_str(), currentDifficulty); - SetParticipantMiningStatus("share found; submitting"); - int submitId = msgId++; - pendingSubmitIds.insert(submitId); - gParticipantSharesSent.fetch_add(1); - sessionSharesSent++; - json submit = {{"id",submitId},{"method","mining.submit"}, - {"params",json::array({username, jobId, ToHexStr(&nNonce,4)})}}; - if (!StratumSendLine(s, submit)) { - LogPrint("worker", "[worker] share submit send FAILED" - " -- pool connection lost\n"); - break; - } - } + // -- Hash rate accounting (hashing happens on the worker threads) -- + hashesSinceSample = gParticipantHashes.load() - sessionStartHashes; + UpdateParticipantHashRate(hashesSinceSample, sampleStart); - nNonce++; - if (nNonce == 0) { - LogPrint("worker", "[worker] nonce space exhausted for job %s" - " -- waiting for new job\n", jobId.c_str()); - haveJob = false; + // -- Submit any shares the hash threads found -- + bool submitFailed = false; + { + std::vector pending; + { + std::lock_guard lk(shareHitsMtx); + pending.swap(shareHits); + } + for (ParticipantShareHit& hit : pending) + { + if (hit.jobId != jobId) + { + LogPrint("worker", "[worker] dropping stale share for old job %s\n", + hit.jobId.c_str()); + continue; + } + LogPrint("worker", "[worker] share found! nonce=%u job=%s difficulty=%.4f\n", + hit.nonce, hit.jobId.c_str(), currentDifficulty); + SetParticipantMiningStatus("share found; submitting"); + int submitId = msgId++; + pendingSubmitIds.insert(submitId); + gParticipantSharesSent.fetch_add(1); + sessionSharesSent++; + json submit = {{"id",submitId},{"method","mining.submit"}, + {"params",json::array({username, hit.jobId, + ToHexStr(&hit.nonce,4)})}}; + if (!StratumSendLine(s, submit)) { + LogPrint("worker", "[worker] share submit send FAILED" + " -- pool connection lost\n"); + submitFailed = true; + break; + } + } } + if (submitFailed) + break; } // ------------------------------------------------------------------ - // Cleanup + // Cleanup: stop the hash threads, then drop the pool connection // ------------------------------------------------------------------ - RandomXDestroyMinerVM(rxvm); + { + std::lock_guard lk(jobShared.mtx); + jobShared.stop = true; + jobShared.haveJob = false; + } + for (std::thread& t : hashThreads) + t.join(); closesocket(s); + hashesSinceSample = gParticipantHashes.load() - sessionStartHashes; UpdateParticipantHashRate(hashesSinceSample, sampleStart); SetParticipantMiningStatus("disconnected"); LogPrint("worker", "[worker] disconnected from pool" diff --git a/src/main.h b/src/main.h index c458412..307cee5 100644 --- a/src/main.h +++ b/src/main.h @@ -84,6 +84,9 @@ extern int nMineMode; // something a previous session wrote. extern bool fMineModeFromCommandLine; extern string strParticipantPool; // participant: pool .btf address +extern string strPayoutAddress; // participant: /payoutaddr override (mine without a wallet) +extern string strPayoutFile; // participant: /payoutfile address list, rotated per reconnect +extern int nHashThreads; // participant: /hashthreads override (0 = cores-1) extern string strPoolName; // operator: announced pool name extern string strPoolDashboardUrl; // operator: optional dashboard URL extern double dPoolFeePercent; // operator: announced fee percent diff --git a/src/main_gui.cpp b/src/main_gui.cpp index 875988f..0b4e091 100644 --- a/src/main_gui.cpp +++ b/src/main_gui.cpp @@ -115,6 +115,9 @@ static void PrintUsage() printf("Mining mode:\n"); printf(" /operator\n"); printf(" /participant=POOL_BTF_ADDRESS\n"); + printf(" /payoutaddr=BTF_ADDRESS (participant: pay shares to this address, no wallet needed)\n"); + printf(" /payoutfile=FILE (participant: rotate payout addresses from file)\n"); + printf(" /hashthreads=N (participant: mining threads, default cores-1)\n"); printf(" /solomine\n"); printf(" /genproclimit=N (mining threads; 0 or absent = every core but one)\n"); printf(" /checkblocks=N (blocks re-verified at startup, default 288, 0 = all)\n"); @@ -225,6 +228,18 @@ static void ParseStartupArguments(int argc, char* argv[]) fMineModeFromCommandLine = true; } + string payoutAddr = argval2(argc, argv, "/payoutaddr", "-payoutaddr"); + if (!payoutAddr.empty()) + strPayoutAddress = payoutAddr; + + string payoutFile = argval2(argc, argv, "/payoutfile", "-payoutfile"); + if (!payoutFile.empty()) + strPayoutFile = payoutFile; + + string hashThreads = argval2(argc, argv, "/hashthreads", "-hashthreads"); + if (!hashThreads.empty()) + nHashThreads = atoi(hashThreads.c_str()); + string poolName = argval2(argc, argv, "/poolname", "-poolname"); if (!poolName.empty()) strPoolName = poolName; @@ -610,6 +625,31 @@ int main(int argc, char* argv[]) ReacceptWalletTransactions(); + // /dumpaddresses=N:FILE -- generate N fresh wallet keys and write their + // addresses (one per line) to FILE, then exit. Keys stay in this wallet; + // the file alone is enough for a miner elsewhere to rotate payout + // addresses via /payoutfile. (No HD/xpub exists in this wallet format.) + string dumpSpec = argval2(argc, argv, "/dumpaddresses", "-dumpaddresses"); + if (!dumpSpec.empty()) + { + size_t colon = dumpSpec.rfind(':'); + if (colon == string::npos) { fprintf(stderr, "usage: /dumpaddresses=N:FILE\n"); return 1; } + int n = atoi(dumpSpec.substr(0, colon).c_str()); + string file = dumpSpec.substr(colon + 1); + FILE* f = fopen(file.c_str(), "w"); + if (!f) { fprintf(stderr, "cannot open %s\n", file.c_str()); return 1; } + for (int i = 0; i < n; i++) + { + CKey key; + key.MakeNewKey(); + if (!AddKey(key)) { fclose(f); fprintf(stderr, "AddKey failed at %d\n", i); return 1; } + fprintf(f, "%s\n", PubKeyToAddress(key.GetPubKey()).c_str()); + } + fclose(f); + printf("wrote %d fresh wallet addresses to %s\n", n, file.c_str()); + return 0; + } + if (!StartNode(strErrors)) { fprintf(stderr,"StartNode: %s\n",strErrors.c_str()); return 1; } if (nMineMode == MINE_OPERATOR) { diff --git a/src/net.cpp b/src/net.cpp index 449d91c..5cf2f8c 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -1054,6 +1054,18 @@ CNode* ConnectNodeBtfResolved(const string& strBtfAddr, const string& strMeeting void ThreadBtfAccept(void* parg) { printf("ThreadBtfAccept started\n"); + // In operator mode the pool server registers this same node identity as a + // stratum service at the rendezvous relays, and a relay only holds ONE + // service registration per pubkey -- peer-accept and pool-accept would + // kick each other off forever, breaking both. The pool's registration is + // the one that must be stable (that's what miners resolve and dial), so + // the node skips inbound rendezvous here and relies on outbound .btf + // connections for block sync. + if (nMineMode == MINE_OPERATOR) + { + printf("ThreadBtfAccept: operator mode -- skipping peer rendezvous registration\n"); + return; + } unsigned char pk[32]; unsigned char sk[32]; while (!BtfGetIdentity(pk, sk)) @@ -1928,11 +1940,29 @@ bool StartNode(string& strError) { int nErr = WSAGetLastError(); if (nErr == WSAEADDRINUSE) - strError = strprintf("Error: Unable to bind to port %s on this computer. The program is probably already running.", addrLocalHost.ToString().c_str()); + { + // Another local instance holds the default port. Fall back to an + // ephemeral port: inbound direct-IP peering is optional, and .btf + // rendezvous connectivity is unaffected by the local listen port. + sockaddr.sin_port = htons(0); + if (bind(hListenSocket, (struct sockaddr*)&sockaddr, sizeof(sockaddr)) == SOCKET_ERROR) + { + strError = strprintf("Error: Unable to bind any port on this computer (bind returned error %d)", WSAGetLastError()); + printf("%s\n", strError.c_str()); + return false; + } + struct sockaddr_in bound; + socklen_t nBoundLen = sizeof(bound); + if (getsockname(hListenSocket, (struct sockaddr*)&bound, &nBoundLen) == 0) + addrLocalHost.port = bound.sin_port; + printf("default port in use by another instance; bound to ephemeral %s instead\n", addrLocalHost.ToString().c_str()); + } else + { strError = strprintf("Error: Unable to bind to port %s on this computer (bind returned error %d)", addrLocalHost.ToString().c_str(), nErr); - printf("%s\n", strError.c_str()); - return false; + printf("%s\n", strError.c_str()); + return false; + } } printf("bound to addrLocalHost = %s\n\n", addrLocalHost.ToString().c_str()); diff --git a/src/nostr.cpp b/src/nostr.cpp index 1fd60ad..21104bc 100644 --- a/src/nostr.cpp +++ b/src/nostr.cpp @@ -92,6 +92,11 @@ const char* pszNostrRelays[] = { "wss://offchain.pub", "wss://nostr.mom", "wss://relay.primal.net", + "wss://nostr.oxtr.dev", + "wss://relay.nos.social", + "wss://nostr.fmt.wiz.biz", + "wss://relay.current.fyi", + "wss://relay.nostr.bg", }; const int nNostrRelays = ARRAYLEN(pszNostrRelays); diff --git a/src/rpc.cpp b/src/rpc.cpp index 41c0328..98b6208 100644 --- a/src/rpc.cpp +++ b/src/rpc.cpp @@ -143,6 +143,20 @@ static bool RebuildJob() return false; } + // Don't issue jobs until the chain is synced: mining against a stale + // tip produces blocks the network will orphan, and payout entries + // that can never be paid (their coinbases never exist on the main + // chain). Require at least one peer and a fresh tip. + if (vNodes.empty()) { + LogPrint("pool", "RebuildJob: no peers yet -- not issuing jobs\n"); + return false; + } + if (GetAdjustedTime() - (int64)pindexBest->nTime > 30*60) { + LogPrint("pool", "RebuildJob: tip is %llds old -- still syncing, not issuing jobs\n", + (long long)(GetAdjustedTime() - (int64)pindexBest->nTime)); + return false; + } + CBlockIndex* pindexPrev = pindexBest; unsigned int nBits = GetNextWorkRequired(pindexPrev); @@ -218,6 +232,8 @@ static bool RebuildJob() struct PendingPayout { int matureAtHeight; std::map amounts; // address -> satoshis + int retries = 0; // exec-thread only: SendMoney attempts so far + int64 retryAfter = 0; // exec-thread only: unix time, 0 = ready now }; static std::vector gPendingPayouts; @@ -403,6 +419,11 @@ static void PayoutThreadFn(void*) } for (auto& pp : batch) { + if (pp.retryAfter > GetTime()) { + std::lock_guard lk(gPayoutExecMutex); + gPayoutExecQueue.push_back(pp); + continue; + } LogPrint("payout", "[payout] executing payments for block matured at %d" " (%zu miners)\n", pp.matureAtHeight, pp.amounts.size()); for (auto& kv : pp.amounts) { @@ -418,13 +439,34 @@ static void PayoutThreadFn(void*) CWalletTx wtx; wtx.mapValue["comment"] = strprintf("Pool payout height %d", pp.matureAtHeight); - if (SendMoney(sc, kv.second, wtx)) + if (SendMoney(sc, kv.second, wtx)) { LogPrint("payout", "[payout] paid %s to %s\n", FormatMoney(kv.second).c_str(), kv.first.c_str()); - else + } else if (pp.retries < 30) { + // Transient failures (immature coinbase, reorg, locked + // wallet) must not silently burn the payment -- requeue + // with backoff. Permanent cases (orphaned coinbase) give + // up loudly after 30 tries (~30 min). + PendingPayout rp; + rp.matureAtHeight = pp.matureAtHeight; + rp.amounts[kv.first] = kv.second; + rp.retries = pp.retries + 1; + rp.retryAfter = GetTime() + 60; + { + std::lock_guard lk(gPayoutExecMutex); + gPayoutExecQueue.push_back(std::move(rp)); + } LogPrint("payout", "[payout] FAILED to pay %s to %s" - " (insufficient funds or wallet locked?)\n", - FormatMoney(kv.second).c_str(), kv.first.c_str()); + " -- retry %d/30 in 60s\n", + FormatMoney(kv.second).c_str(), kv.first.c_str(), + pp.retries + 1); + } else { + LogPrint("payout", "[payout] GIVING UP on %s to %s after %d" + " retries (orphaned coinbase or persistent wallet" + " error?) -- payment forfeited\n", + FormatMoney(kv.second).c_str(), kv.first.c_str(), + pp.retries); + } } } @@ -543,6 +585,45 @@ static void PushReady(btf_socket_t s) gReadyQueue.push_back(s); } +// Loopback stratum listener: lets a miner running on the SAME machine reach +// this pool over plain TCP (127.0.0.1:8334), skipping the .btf rendezvous +// entirely. The rendezvous relay only allows one service registration per +// pubkey, so the pool service and the node's own peer-accept fight over the +// node identity at the relay -- loopback avoids that fight for local miners. +static void LocalAcceptFn(void*) +{ + SOCKET ls = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (ls == INVALID_SOCKET) + return; + int yes = 1; + setsockopt(ls, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes)); + struct sockaddr_in addr; + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = htons(8334); + if (bind(ls, (struct sockaddr*)&addr, sizeof(addr)) != 0 || listen(ls, 16) != 0) + { + LogPrint("pool", "[pool] loopback stratum bind/listen failed on 127.0.0.1:8334\n"); + closesocket(ls); + return; + } + LogPrint("pool", "[pool] loopback stratum listening on 127.0.0.1:8334\n"); + while (!fShutdown && gPoolRunning) + { + SOCKET c = accept(ls, NULL, NULL); + if (c == INVALID_SOCKET) + { + if (fShutdown || !gPoolRunning) + break; + continue; + } + LogPrint("pool", "[pool] miner arrived via loopback -- pushing to event loop\n"); + PushReady((btf_socket_t)c); + } + closesocket(ls); +} + // --------------------------------------------------------------------------- // I/O helpers -- event loop thread only // --------------------------------------------------------------------------- @@ -1307,6 +1388,11 @@ static void AcceptOneFn(void* arg) LogPrint("pool", "[pool] accept thread started for relay %s\n", relay.c_str()); while (!fShutdown && gPoolRunning) { + // Advertise this relay as the pool's meeting node BEFORE registering: + // RvServiceRegister blocks until a miner pairs, and miners can only + // find us if the published descriptor already names a relay where the + // pool (not the node's peer-accept) holds the registration. + BtfSetActiveRelay(relay); // Blocks until a miner connects to us at this relay btf::RvSocket rv = btf::RvServiceRegister(host.c_str(), (unsigned short)port, pk); @@ -1408,6 +1494,7 @@ void ThreadRPCServer(void*) LoadPendingPayouts(); _beginthread(AcceptLoopFn, 0, NULL); + _beginthread(LocalAcceptFn, 0, NULL); _beginthread(ThreadBtfPoolAnnouncer, 0, NULL); _beginthread(PayoutThreadFn, 0, NULL); -- cgit v1.2.3