net: fail over a stalled block fetch to another peer

The 15s retry spacing only helps when several peers announce the block:
the queued retries are created by inv handling, so a block announced by
a single peer that then goes quiet was re-requested from nobody, and the
tip stayed frozen until somebody happened to re-announce (#33).

Track every block getdata in mapBlockFetch, erase the entry when the
body arrives, and let SendMessages re-issue a request that has gone
unanswered for BLOCK_FETCH_RETRY_SECS to the next peer that has not had
a turn this round, restarting the round once every connected peer has.
Client peers are skipped -- they declared they cannot serve bodies.
Tracking is capped at 8 in-flight blocks so initial download, where
hundreds are legitimately outstanding to one peer, is not duplicated to
everybody else, and an entry is dropped after 10 minutes so a hash that
was announced and never existed cannot be chased forever.

The failover log line is deliberately not behind a debug category: it
firing means a peer announces blocks it does not serve, which is the
pathology being diagnosed.

1 files changed, 105 insertions, 0 deletions

diff --git a/src/main.cpp b/src/main.cpp
index 7b3e555..d9fe713 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -64,6 +64,31 @@ CBlockIndex* pindexBest = NULL;
 map<uint256, CBlock*> mapOrphanBlocks;
 multimap<uint256, CBlock*> mapOrphanBlocksByPrev;
 
+// Block bodies we have sent a getdata for and not yet received, so the
+// failover pass in SendMessages can tell "requested and being served" from
+// "requested and gone quiet" and re-issue the request to a different peer
+// (#33). Keyed by block hash; setAsked is the peers already tried since the
+// round last restarted, so the retry rotates instead of hammering the
+// announcer that is not serving. Guarded by cs_main, which both the message
+// handlers and SendMessages already hold.
+//
+// Capped small on purpose. At the tip there is one block in flight; during
+// initial download there are legitimately hundreds outstanding to one peer,
+// and rotating those would duplicate the whole download to every other peer.
+// Anything past the cap is simply not tracked and behaves as before.
+struct CBlockFetch
+{
+    int64 nTimeFirstAsked;  // when the first getdata for this block went out
+    int64 nTimeAsked;       // when the latest getdata went out
+    set<CAddress> setAsked; // peers tried since the round last restarted
+};
+map<uint256, CBlockFetch> mapBlockFetch;
+static const int BLOCK_FETCH_TRACK_MAX = 8;
+// A block nobody has served for this long is not a fetch problem: the hash may
+// have been announced and never existed. Drop the entry rather than getdata
+// forever; a re-announcement recreates it.
+static const int BLOCK_FETCH_GIVEUP_SECS = 10 * 60;
+
 map<uint256, CDataStream*> mapOrphanTransactions;
 multimap<uint256, CDataStream*> mapOrphanTransactionsByPrev;
 
@@ -3268,6 +3293,12 @@ bool ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv)
         CInv inv(MSG_BLOCK, pblock->GetHash());
         pfrom->AddInventoryKnown(inv);
 
+        // The body is here, so the fetch is over whatever validation says
+        // next -- unlike the mapAlreadyAskedFor erase below, which stays
+        // conditional on acceptance so a peer feeding us garbage does not
+        // cancel the scheduled retries for the real thing.
+        mapBlockFetch.erase(inv.hash);
+
         if (ProcessBlock(pfrom, pblock.release()))
         {
             mapAlreadyAskedFor.erase(inv);
@@ -3396,12 +3427,86 @@ bool SendMessages(CNode* pto)
             const CInv& inv = (*pto->mapAskFor.begin()).second;
             if (LogAcceptsCategory("net")) printf("sending getdata: %s\n", inv.ToString().c_str());
             if (!AlreadyHave(txdb, inv))
+            {
                 vAskFor.push_back(inv);
+                // Blocks get their delivery tracked so the failover pass
+                // below can chase the ones that go quiet. See mapBlockFetch
+                // for why anything past the cap is left untracked.
+                if (inv.type == MSG_BLOCK &&
+                    (mapBlockFetch.count(inv.hash) || mapBlockFetch.size() < BLOCK_FETCH_TRACK_MAX))
+                {
+                    CBlockFetch& fetch = mapBlockFetch[inv.hash];
+                    if (fetch.nTimeFirstAsked == 0)
+                        fetch.nTimeFirstAsked = GetTime();
+                    fetch.nTimeAsked = GetTime();
+                    fetch.setAsked.insert(pto->addr);
+                }
+            }
             pto->mapAskFor.erase(pto->mapAskFor.begin());
         }
         if (!vAskFor.empty())
             pto->PushMessage("getdata", vAskFor);
 
+
+        //
+        // Message: getdata (block fetch failover)
+        //
+        // A requested block that has not arrived in BLOCK_FETCH_RETRY_SECS is
+        // not still in transit -- a body is at most 1 MB and somebody
+        // demonstrably has it -- so ask somebody else instead of waiting on
+        // whoever was asked first (#33). Each peer the message handler visits
+        // takes at most one turn per block per round; once every connected
+        // peer has had its turn and a further interval passes with nobody
+        // new to ask, the round restarts. A peer that declared itself a
+        // client cannot serve block bodies and is no use as a target.
+        if (!pto->fClient)
+        {
+            vector<CInv> vFailover;
+            int64 nNowSecs = GetTime();
+            for (map<uint256, CBlockFetch>::iterator mi = mapBlockFetch.begin(); mi != mapBlockFetch.end();)
+            {
+                CBlockFetch& fetch = (*mi).second;
+                CInv inv(MSG_BLOCK, (*mi).first);
+                if (AlreadyHave(txdb, inv))
+                {
+                    // Arrived by some other road -- our own miner, an orphan
+                    // parked in mapOrphanBlocks. Nothing left to chase.
+                    mapBlockFetch.erase(mi++);
+                    continue;
+                }
+                if (nNowSecs - fetch.nTimeFirstAsked > BLOCK_FETCH_GIVEUP_SECS)
+                {
+                    printf("block fetch: giving up on %s -- no peer served it in %d s\n",
+                           inv.ToString().c_str(), (int)(nNowSecs - fetch.nTimeFirstAsked));
+                    mapBlockFetch.erase(mi++);
+                    continue;
+                }
+                if (nNowSecs - fetch.nTimeAsked >= BLOCK_FETCH_RETRY_SECS)
+                {
+                    // The handler loop visits every peer within ~100ms, so a
+                    // request stale by two full intervals means nobody outside
+                    // setAsked was there to ask. Start the round over.
+                    if (nNowSecs - fetch.nTimeAsked >= 2 * BLOCK_FETCH_RETRY_SECS)
+                        fetch.setAsked.clear();
+                    if (!fetch.setAsked.count(pto->addr))
+                    {
+                        // Deliberately not behind a debug category: this line
+                        // firing is a peer announcing blocks it does not
+                        // serve, which is the pathology of #33.
+                        printf("block fetch failover: %s unanswered for %d s, re-requesting from %s\n",
+                               inv.ToString().c_str(), (int)(nNowSecs - fetch.nTimeAsked),
+                               pto->addr.ToString().c_str());
+                        vFailover.push_back(inv);
+                        fetch.nTimeAsked = nNowSecs;
+                        fetch.setAsked.insert(pto->addr);
+                    }
+                }
+                ++mi;
+            }
+            if (!vFailover.empty())
+                pto->PushMessage("getdata", vFailover);
+        }
+
     }
     return true;
 }