net: do not consume a header until the body is there, and cap what one can claim

Upstream #109's message-stall rework, hand-merged so it keeps the progress
timing this tree already had.

The old path consumed the header, found the payload short, inserted the header
back at the front of vRecv, then Sleep(100) inside the shared message thread.
A peer that announces a large payload and drips bytes made every pass pay an
O(n) insert and a 100 ms stall -- one peer slowing the thread that serves all
of them. Now the header is read from a copy and only skipped once the whole
body has arrived, so there is nothing to put back and nothing to sleep for.

Kept OURS, not upstream's, on the timeout: upstream restarts its clock only
when the (size, command) pair changes, which drops a peer sending one big
message slowly but honestly. This times the STALL -- the clock restarts on any
progress -- which is the distinction that matters on a slow link.

Also from #109: a header could announce 256 MB while serialize.h refuses
anything over MAX_SIZE (32 MB), so the check admitted an allocation the parser
would reject anyway and burned a peer slot for two minutes doing it.
MAX_PROTOCOL_MESSAGE_SIZE ties the two together. 8x less a peer can make us
hold, and this node takes public Stratum.

And #103's other half: ~CNode closed its socket with a raw closesocket(),
invisible to the accounting, and left hSocket set afterwards -- which is how a
descriptor number that now belongs to somebody else gets closed a second time.

(cherry picked from commit 2f25e2e07985e9bc732cadfbb3c9ac233895a814)

1 files changed, 60 insertions, 22 deletions

diff --git a/src/main.cpp b/src/main.cpp
index 37a15c6..11eb11d 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -2827,11 +2827,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);
@@ -2841,7 +2843,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;
         }
@@ -2852,30 +2857,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);