1 files changed, 439 insertions, 122 deletions

diff --git a/src/main.cpp b/src/main.cpp
index f83164f..252911b 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -87,6 +87,8 @@ CKey keyUser;
 string strSetDataDir;
 int nDropMessagesTest = 0;
 bool fSoloMineTest = false; // /solomine: mine without requiring a peer (local test)
+bool fPoolNoAnnounce = false; // /noannounce: operator serves miners but never publishes pool announcements
+bool fPoolMineBlind = false; // /poolmineblind: issue jobs even when no peer has reported a height
 
 // Settings
 int fGenerateBitcoins;
@@ -102,6 +104,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;
 string strPoolStatusFile;
@@ -163,12 +168,32 @@ int MinerThreadCount()
 {
     if (nMinerThreads > 0)
         return nMinerThreads;
-    unsigned int nCores = std::thread::hardware_concurrency();
+    unsigned int nCores = AvailableCoreCount();
     if (nCores <= 1)
         return 1;
     return (int)(nCores - 1);
 }
 
+// Hash threads one participant session runs.
+//
+// Explicit /hashthreads=N is honored exactly, including 1 -- a miner who says
+// one means one. Automatic leaves a core to the node, the Stratum I/O loop and
+// everything else the process still has to do, for the same reason
+// MinerThreadCount() does.
+//
+// Lives here, next to that function, because the two are easy to confuse and
+// StartMinerThreads() needs to be able to tell them apart: only ONE session
+// runs in participant mode, and this is what sizes it.
+int ParticipantHashThreadCount()
+{
+    if (nHashThreads > 0)
+        return nHashThreads;
+    int nWorkers = (int)AvailableCoreCount();
+    if (nWorkers < 2)
+        nWorkers = 2;
+    return nWorkers - 1;
+}
+
 static void SecureClearBytes(vector<unsigned char>& vch);
 static void SecureClearKeyingMaterial(CKeyingMaterial& vch);
 
@@ -1551,7 +1576,7 @@ int CMerkleTx::GetBlocksToMaturity() const
 {
     if (!IsCoinBase())
         return 0;
-    return max(0, (COINBASE_MATURITY+20) - GetDepthInMainChain());
+    return max(0, COINBASE_SPENDABLE_DEPTH - GetDepthInMainChain());
 }
 
 
@@ -2819,9 +2844,10 @@ bool AlreadyHave(CTxDB& txdb, const CInv& inv)
     {
     case MSG_TX:        return mapTransactions.count(inv.hash) || txdb.ContainsTx(inv.hash);
     case MSG_BLOCK:     return mapBlockIndex.count(inv.hash) || mapOrphanBlocks.count(inv.hash);
-    case MSG_REVIEW:    return true;
-    case MSG_PRODUCT:   return mapProducts.count(inv.hash);
     }
+    // MSG_REVIEW / MSG_PRODUCT / MSG_TABLE are the 2009 marketplace inventory
+    // types. Nothing announces or serves them any more (see ProcessMessage),
+    // so they fall through to the default and are never requested.
     // Don't know what it is, just say we already got one
     return true;
 }
@@ -2864,11 +2890,13 @@ bool ProcessMessages(CNode* pfrom)
             if (LogAcceptsCategory("net")) printf("\n\nPROCESSMESSAGE SKIPPED %d BYTES\n\n", (int)(pstart - vRecv.begin()));
         vRecv.erase(vRecv.begin(), pstart);
 
-        // Read the header from a copy first. The old path consumed the header,
-        // noticed the payload was incomplete, inserted the header back at the
-        // front of the receive buffer, then slept inside the shared message
-        // thread. A peer could announce a large payload and drip bytes forever,
-        // making every pass pay for an O(n) insert and a 100 ms stall.
+        // Read the header from a copy first, and do not consume it until the
+        // whole body is here. The old path consumed the header, noticed the
+        // payload was incomplete, inserted the header back at the front of the
+        // receive buffer, then slept inside the shared message thread. A peer
+        // could announce a large payload and drip bytes forever, making every
+        // pass pay for an O(n) insert and a 100 ms stall -- a cheap way for one
+        // peer to slow the thread that serves all of them. (upstream #109)
         CMessageHeader hdr;
         CDataStream vHeader(vRecv.begin(), vRecv.begin() + sizeof(CMessageHeader),
                             vRecv.nType, vRecv.nVersion);
@@ -2878,7 +2906,10 @@ bool ProcessMessages(CNode* pfrom)
             if (LogAcceptsCategory("net")) printf("\n\nPROCESSMESSAGE: ERRORS IN HEADER %s\n\n\n", hdr.GetCommand().c_str());
             if (hdr.nMessageSize > MAX_PROTOCOL_MESSAGE_SIZE)
                 pfrom->fDisconnect = true;
-            pfrom->nIncompleteMessageStart = 0;
+            pfrom->nMsgWaitSince = 0;
+            pfrom->nMsgWaitSize  = 0;
+            // The header was read from a copy, so it is still in vRecv. Skip it
+            // explicitly or this loop never advances.
             vRecv.ignore(sizeof(CMessageHeader));
             continue;
         }
@@ -2889,30 +2920,63 @@ bool ProcessMessages(CNode* pfrom)
         unsigned int nPayloadAvailable = vRecv.size() - sizeof(CMessageHeader);
         if (nMessageSize > nPayloadAvailable)
         {
-            if (pfrom->nIncompleteMessageStart == 0 ||
-                pfrom->nIncompleteMessageSize != nMessageSize ||
-                pfrom->strIncompleteMessageCommand != strCommand)
+            // Wait for the rest of the message. The header stays in vRecv.
+            //
+            // This is the mechanism the original note asked for: "need a
+            // mechanism to give up waiting for overlong message size error".
+            // Without one the wait is unbounded, and unbounded here means
+            // permanent. The header stays at the front of vRecv and the loop
+            // breaks; if the body never completes, every later pass re-reads
+            // that same header and breaks in the same place. The peer is never
+            // parsed again -- no blocks, no inv, not even its version message,
+            // so it never reports a height either, which is why a wedged node
+            // has peers and GetPeerMedianHeight() still returns -1.
+            //
+            // Nothing rescues it. A header could claim up to 256 MB and still
+            // pass CMessageHeader::IsValid() -- now capped at
+            // MAX_PROTOCOL_MESSAGE_SIZE -- and any byte that arrives
+            // refreshes nLastRecv, so the inactivity timeout that drops a
+            // silent peer never fires on one that is merely unreadable. The
+            // socket stays ESTABLISHED and select() keeps offering it.
+            // Measured on a wedged pool: this branch ran 2014 times in the
+            // deaf window, against 607 in the healthy window before it.
+            if (LogAcceptsCategory("net")) printf("MESSAGE-BREAK 2\n");
+            //
+            // Time the STALL, not the wait. A big message on a slow link is
+            // still arriving, and dropping that peer would be a bug of our own
+            // making; what must be caught is a buffer that has stopped growing
+            // while we go on expecting more. So the clock restarts whenever
+            // even one more byte has landed since the last look.
+            //
+            // Upstream #109 times this differently: it restarts the clock only
+            // when the (size, command) pair changes, which drops a peer sending
+            // one big message slowly but honestly. Progress is the better test,
+            // so that half is kept as it was.
+            int64 nNow = GetTime();
+            if (pfrom->nMsgWaitSince == 0 || nPayloadAvailable > pfrom->nMsgWaitSize)
             {
-                pfrom->nIncompleteMessageStart = GetTime();
-                pfrom->nIncompleteMessageSize = nMessageSize;
-                pfrom->strIncompleteMessageCommand = strCommand;
+                pfrom->nMsgWaitSince = nNow;
+                pfrom->nMsgWaitSize  = nPayloadAvailable;
             }
-            else if (GetTime() - pfrom->nIncompleteMessageStart >
-                     BTF_INCOMPLETE_MESSAGE_TIMEOUT_SECS)
+            else if (nNow - pfrom->nMsgWaitSince > BTF_MSG_WAIT_MAX_SECS)
             {
-                LogPrint("net",
-                         "disconnecting %s: incomplete %s message, %u of %u bytes after %d seconds\n",
-                         pfrom->addr.ToString().c_str(), strCommand.c_str(),
-                         nPayloadAvailable, nMessageSize,
-                         BTF_INCOMPLETE_MESSAGE_TIMEOUT_SECS);
+                // Deliberately not behind a debug category. This is the node
+                // going deaf to a peer, which is the thing nobody could see.
+                printf("peer %s announced a %u byte '%s' and has sent nothing more "
+                       "for %d s (%d of %u payload bytes) -- dropping, this peer is unreadable\n",
+                       pfrom->addr.ToString().c_str(), nMessageSize,
+                       strCommand.c_str(), (int)(nNow - pfrom->nMsgWaitSince),
+                       (int)nPayloadAvailable, nMessageSize);
                 pfrom->fDisconnect = true;
             }
+            // No re-insert, no Sleep(): the header was never consumed, so there
+            // is nothing to put back, and sleeping here stalls every peer.
             break;
         }
-        pfrom->nIncompleteMessageStart = 0;
-        pfrom->nIncompleteMessageSize = 0;
-        pfrom->strIncompleteMessageCommand.clear();
         vRecv.ignore(sizeof(CMessageHeader));
+        // The body arrived, so this peer is readable again.
+        pfrom->nMsgWaitSince = 0;
+        pfrom->nMsgWaitSize  = 0;
 
         // Copy message to its own buffer
         CDataStream vMsg(vRecv.begin(), vRecv.begin() + nMessageSize, vRecv.nType, vRecv.nVersion);
@@ -2941,7 +3005,6 @@ bool ProcessMessages(CNode* pfrom)
 
 bool ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
 {
-    static map<unsigned int, vector<unsigned char> > mapReuseKey;
     if (LogAcceptsCategory("net")) { printf("received: %-12s (%d bytes)  ", strCommand.c_str(), (int)vRecv.size()); for (int i = 0; i < min(vRecv.size(), (unsigned int)25); i++) printf("%02x ", vRecv[i] & 0xff); printf("\n"); }
     if (nDropMessagesTest > 0 && GetRand(nDropMessagesTest) == 0)
     {
@@ -3180,22 +3243,18 @@ bool ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
     }
 
 
-    else if (strCommand == "review")
-    {
-        CDataStream vMsg(vRecv);
-        CReview review;
-        vRecv >> review;
-
-        CInv inv(MSG_REVIEW, review.GetHash());
-        pfrom->AddInventoryKnown(inv);
-
-        if (review.AcceptReview())
-        {
-            // Relay the original message as-is in case it's a higher version than we know how to parse
-            RelayMessage(inv, vMsg);
-            mapAlreadyAskedFor.erase(inv);
-        }
-    }
+    // "review" used to be handled here. It took a CReview off the wire from an
+    // unauthenticated peer, checked a signature the sender had made with a key
+    // of its own choosing -- which proves possession of that key and authorises
+    // nothing -- and then appended the message to reviews.dat by reading the
+    // whole vector for the target hash, pushing one entry, and writing all of
+    // it back. That is quadratic disk for a stream an attacker controls, on a
+    // record whose mapValue is an unbounded map<string,string> read straight
+    // from the socket, followed by AddAtomsAndPropagate walking a link graph
+    // the same peer supplied, and finally RelayMessage forwarding it to every
+    // other node. Unbounded remote disk growth with network amplification and
+    // no authorisation anywhere in the path. The marketplace it belonged to has
+    // not existed since 2009; the handler is gone rather than bounded.
 
 
     else if (strCommand == "block")
@@ -3210,41 +3269,46 @@ bool ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
         pfrom->AddInventoryKnown(inv);
 
         if (ProcessBlock(pfrom, pblock.release()))
+        {
             mapAlreadyAskedFor.erase(inv);
-    }
-
 
-    else if (strCommand == "checkorder" || strCommand == "submitorder")
-    {
-        // Legacy Bitcoin 0.1 marketplace messages are not part of Bitflash's
-        // supported protocol. They used to let a remote peer make this wallet
-        // reserve a key or import/broadcast a wallet transaction. Keep the
-        // command names explicit so old peers get a quiet no-op instead of a
-        // wallet side effect.
-        if (LogAcceptsCategory("net"))
-            printf("ProcessMessage(%s) : disabled legacy order message\n", strCommand.c_str());
-        vRecv.clear();
+            // A peer that just handed us a block is demonstrably at least at
+            // that block's height. Without this, nStartingHeight is a snapshot
+            // taken once during the version handshake and never touched again,
+            // so on a long-lived connection it is the peer's height from
+            // however many hours ago it happened to connect. The pool's
+            // behind-the-network check compares against the median of those
+            // numbers, so as connections age it drifts toward "everyone is
+            // behind us" and quietly stops firing -- the opposite of its job.
+            map<uint256, CBlockIndex*>::iterator mi = mapBlockIndex.find(inv.hash);
+            if (mi != mapBlockIndex.end() && mi->second)
+                pfrom->nStartingHeight = max(pfrom->nStartingHeight, mi->second->nHeight);
+        }
     }
 
 
-    else if (strCommand == "reply")
-    {
-        uint256 hashReply;
-        vRecv >> hashReply;
-
-        CRequestTracker tracker;
-        CRITICAL_BLOCK(pfrom->cs_mapRequests)
-        {
-            map<uint256, CRequestTracker>::iterator mi = pfrom->mapRequests.find(hashReply);
-            if (mi != pfrom->mapRequests.end())
-            {
-                tracker = (*mi).second;
-                pfrom->mapRequests.erase(mi);
-            }
-        }
-        if (!tracker.IsNull())
-            tracker.fn(tracker.param1, vRecv);
-    }
+    // "checkorder", "submitorder" and "reply" -- the 2009 buy-from-a-merchant
+    // exchange -- used to be handled here. All three are gone.
+    //
+    // "checkorder" called GenerateNewKey() for any peer that asked, which is
+    // AddKey() and a wallet.dat write, and handed the pubkey back. The comment
+    // said "keep giving the same key to the same ip until they use it", and on
+    // a 2009 network of routable IPs that held. It does not hold here: every
+    // inbound peer arrives through a rendezvous tunnel and is tagged with
+    // BtfMarkerAddr() over three random bytes (net.cpp), so addr.ip is freshly
+    // random per connection and the cache never hits. Reconnect in a loop and
+    // the node mints a key per connection -- unbounded wallet growth, unbounded
+    // growth of the static map that held them, and every one of those keys
+    // missing from every backup taken beforehand, which for this pre-BIP32
+    // wallet is how coins are lost.
+    //
+    // "submitorder" took a CWalletTx from an unauthenticated peer and, if it
+    // was merely a valid transaction, AddToWallet()'d it. Valid is not the same
+    // as ours: any peer could write chosen transactions into wallet.dat.
+    //
+    // "reply" dispatched to a CRequestTracker callback. Nothing in this tree
+    // has ever called PushRequest, so mapRequests was permanently empty and the
+    // handler could only ever be a parser for hostile input.
 
 
     else if (strCommand == "ping")
@@ -3502,6 +3566,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()
 {
     // ------------------------------------------------------------------
@@ -3516,6 +3676,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());
@@ -3550,11 +3755,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",
@@ -3564,11 +3769,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))
@@ -3606,16 +3848,14 @@ static bool PoolParticipantMiner()
     // ------------------------------------------------------------------
     if (!RandomXFastReady()) {
         SetParticipantMiningStatus("building RandomX dataset (~2GB, one-time)");
-        unsigned int nThreads = std::thread::hardware_concurrency();
+        // Fill with the cores we are permitted to use, not the ones the host
+        // happens to have. More threads than the cpuset allows does not build
+        // the dataset any faster -- the CPU budget is the same either way --
+        // it just oversubscribes, and this runs alongside the pool node.
+        unsigned int nThreads = AvailableCoreCount();
         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");
 
     // ------------------------------------------------------------------
@@ -3633,7 +3873,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;
     }
 
@@ -3643,11 +3882,31 @@ 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.
+    // ------------------------------------------------------------------
+    // One count, decided in one place (main.cpp:MinerThreadCount's neighbour),
+    // so that what StartMinerThreads() reports at startup and what actually
+    // hashes here can never drift apart. Explicit /hashthreads=N is honored
+    // exactly, including 1; automatic leaves one core for node + I/O threads.
+    int nWorkers = ParticipantHashThreadCount();
+    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;
@@ -3664,6 +3923,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");
 
@@ -3771,6 +4032,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);
@@ -3826,46 +4095,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"
@@ -4090,7 +4384,7 @@ bool BitcoinMiner(int nThreadId)
     // Build the fast (~2 GB dataset) RandomX mode once, up front, so we're
     // not stuck mining in the much slower light/cache-only mode forever.
     if (!RandomXFastReady()) {
-        unsigned int nThreads = std::thread::hardware_concurrency();
+        unsigned int nThreads = AvailableCoreCount();
         RandomXInitDataset(nThreads > 0 ? (int)nThreads : 1);
     }
 
@@ -4177,7 +4471,19 @@ bool BitcoinMiner(int nThreadId)
             vector<char> vfAlreadyAdded(mapTransactions.size());
             bool fFoundSomething = true;
             unsigned int nBlockSize = 0;
-            while (fFoundSomething && nBlockSize < MAX_SIZE/2)
+            // Assembly must stop below the consensus cap that CheckBlock()
+            // enforces, or we mine a block our own node then rejects.
+            //
+            // This used to read MAX_SIZE/2 (16 MB), which was consistent while
+            // CheckBlock() also tested against MAX_SIZE (32 MB). Adding
+            // MAX_BLOCK_SIZE inverted it: assembly could fill 16 MB and
+            // validation would refuse anything over 1 MB, so a full mempool
+            // meant solving a block and throwing it away -- with this pool at
+            // ~90% of hashrate, that is the network stalling, not just us.
+            //
+            // Half the cap, keeping the original margin: the test runs before
+            // a transaction is added, so the loop can overshoot by one tx.
+            while (fFoundSomething && nBlockSize < MAX_BLOCK_SIZE/2)
             {
                 fFoundSomething = false;
                 unsigned int n = 0;
@@ -4778,7 +5084,18 @@ bool SendMoney(CScript scriptPubKey, int64 nValue, CWalletTx& wtxNew)
         {
             string strError;
             if (nValue + nFeeRequired > GetBalance())
-                strError = strprintf("Error: This is an oversized transaction that requires a transaction fee of %s ", FormatMoney(nFeeRequired).c_str());
+                // Not an oversized transaction -- this branch is simply "the
+                // spendable balance is too small". Saying otherwise (with a fee
+                // of 0.00, because CreateTransaction failed before setting one)
+                // sends anyone reading the log after a failed payout hunting a
+                // transaction-size problem that does not exist.
+                strError = strprintf("Error: insufficient spendable balance:"
+                                     " need %s (plus fee %s) but only %s is spendable"
+                                     " -- mined coin counts as zero until it is %d deep",
+                                     FormatMoney(nValue).c_str(),
+                                     FormatMoney(nFeeRequired).c_str(),
+                                     FormatMoney(GetBalance()).c_str(),
+                                     COINBASE_SPENDABLE_DEPTH);
             else
                 strError = "Error: Transaction creation failed ";
             return error("SendMoney() : %s\n", strError.c_str());