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)

1 files changed, 273 insertions, 45 deletions

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<ParticipantShareHit>* 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<std::mutex> 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<std::mutex> 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
+    //   <name>.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<std::string> 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<unsigned char> 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,11 +3803,30 @@ 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<ParticipantShareHit> shareHits;
+    std::mutex shareHitsMtx;
+    std::vector<std::thread> 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
     // ------------------------------------------------------------------
     std::string jobId;
@@ -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<std::mutex> 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<ParticipantShareHit> pending;
+            {
+                std::lock_guard<std::mutex> 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<std::mutex> 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"