4 files changed, 287 insertions, 30 deletions

diff --git a/relay/btfrv.cpp b/relay/btfrv.cpp
index b1eee55..833c47c 100644
--- a/relay/btfrv.cpp
+++ b/relay/btfrv.cpp
@@ -31,6 +31,7 @@ typedef int SOCKET;
 #include <cstring>
 #include <cstdio>
 #include <string>
+#include <deque>
 #include <map>
 #include <thread>
 #include <mutex>
@@ -44,7 +45,24 @@ static std::atomic<bool> g_stop(false);
 static SOCKET            g_listen = INVALID_SOCKET;
 
 // Services that have registered and are waiting for a client, keyed by pubkey.
-static std::map<std::string, SOCKET> g_waiting;
+// More than one registration per node, on purpose.
+//
+// This used to be one socket per pubkey, and a dial *consumed* it: the first
+// caller took the registration and everyone else was told "not found" until the
+// node noticed and registered again. A node re-registers only after its pairing
+// has been handed off, so between the two there is a window where it is
+// unreachable to the whole network. Measured from a node's own counters: half
+// of all outbound dials failed that way -- 15 of 30 in eleven minutes -- while
+// descriptors resolved 53 times out of 54 and the node's own registrations
+// paired 11 times out of 12. The rendezvous was not losing anybody; it was
+// letting one caller in at a time.
+//
+// A queue per key lets a node keep spare registrations parked, so a dial that
+// arrives while another is being paired finds the next one instead of a closed
+// door. Old nodes register exactly as before and simply keep one entry: the
+// wire protocol does not change, so a relay running this serves both.
+static const size_t MAX_WAITING_PER_KEY = 4;
+static std::map<std::string, std::deque<SOCKET> > g_waiting;
 static std::mutex                    g_mtx;
 
 // How often the reaper walks the waiting map.
@@ -150,16 +168,24 @@ static void ReapDeadRegistrations()
     int nBefore = 0, nReaped = 0;
     {
         std::lock_guard<std::mutex> lk(g_mtx);
-        nBefore = (int)g_waiting.size();
-        std::map<std::string, SOCKET>::iterator it = g_waiting.begin();
+        std::map<std::string, std::deque<SOCKET> >::iterator it = g_waiting.begin();
         while (it != g_waiting.end())
         {
-            if (RegistrationDead(it->second))
+            std::deque<SOCKET>& q = it->second;
+            nBefore += (int)q.size();
+            for (std::deque<SOCKET>::iterator jt = q.begin(); jt != q.end();)
             {
-                CLOSESOCK(it->second);
-                g_waiting.erase(it++);
-                nReaped++;
+                if (RegistrationDead(*jt))
+                {
+                    CLOSESOCK(*jt);
+                    jt = q.erase(jt);
+                    nReaped++;
+                }
+                else
+                    ++jt;
             }
+            if (q.empty())
+                g_waiting.erase(it++);
             else
                 ++it;
         }
@@ -209,34 +235,57 @@ static void HandleIncoming(SOCKET s)
     {
         // Register and wait; a future client will pair us.
         std::lock_guard<std::mutex> lk(g_mtx);
-        // If a stale service for this key exists, drop it.
-        std::map<std::string, SOCKET>::iterator it = g_waiting.find(pubkey);
-        if (it != g_waiting.end()) CLOSESOCK(it->second);
-        g_waiting[pubkey] = s;
-        // leave the socket open in the map; do not close here.
+        std::deque<SOCKET>& q = g_waiting[pubkey];
+
+        // Drop registrations this node has already lost the far end of, rather
+        // than dropping the previous one on principle. The old code closed the
+        // prior registration on every new one, which is why a node could never
+        // hold a spare.
+        for (std::deque<SOCKET>::iterator it = q.begin(); it != q.end();)
+        {
+            if (RegistrationDead(*it)) { CLOSESOCK(*it); it = q.erase(it); }
+            else ++it;
+        }
+
+        // A ceiling, because a stranger can open these. Oldest goes first: it
+        // is the one most likely to be stale.
+        while (q.size() >= MAX_WAITING_PER_KEY)
+        {
+            CLOSESOCK(q.front());
+            q.pop_front();
+        }
+        q.push_back(s);
+        // leave the socket open in the queue; do not close here.
     }
     else if (role == 'C')
     {
         SOCKET svc = INVALID_SOCKET;
         {
             std::lock_guard<std::mutex> lk(g_mtx);
-            std::map<std::string, SOCKET>::iterator it = g_waiting.find(pubkey);
+            std::map<std::string, std::deque<SOCKET> >::iterator it = g_waiting.find(pubkey);
             if (it != g_waiting.end())
             {
-                // Check before handing it over. Pairing a client with a dead
-                // registration produces a tunnel that fails after both sides
-                // think they succeeded, which is the failure the dialling node
-                // reports as "tunnel to X failed" with nothing to explain it.
-                if (RegistrationDead(it->second))
+                std::deque<SOCKET>& q = it->second;
+                // Take the oldest one that is still alive, discarding the dead
+                // as we go. Checking before handing it over matters: pairing a
+                // client with a dead registration produces a tunnel that fails
+                // after both sides think they succeeded, which is the failure
+                // the dialling node reports as "tunnel to X failed" with
+                // nothing to explain it.
+                while (!q.empty())
                 {
-                    CLOSESOCK(it->second);
-                    g_waiting.erase(it);
+                    SOCKET candidate = q.front();
+                    q.pop_front();
+                    if (RegistrationDead(candidate))
+                    {
+                        CLOSESOCK(candidate);
+                        continue;
+                    }
+                    svc = candidate;
+                    break;
                 }
-                else
-                {
-                    svc = it->second;
+                if (q.empty())
                     g_waiting.erase(it);
-                }
             }
         }
         if (svc == INVALID_SOCKET)
diff --git a/src/net.cpp b/src/net.cpp
index 7091981..449d91c 100644
--- a/src/net.cpp
+++ b/src/net.cpp
@@ -51,6 +51,103 @@ int64 nBlocksReceived      = 0;
 int64 nBlocksWithoutParent = 0;
 static int64 nNodeStartTime = 0;
 
+static CCriticalSection cs_btfChurn;
+static int64 nBtfResolveAttempts = 0;
+static int64 nBtfResolveOk = 0;
+static int64 nBtfResolveMiss = 0;
+static int64 nBtfDialAttempts = 0;
+static int64 nBtfDialOk = 0;
+static int64 nBtfDialFailed = 0;
+static int64 nBtfRegisterOk = 0;
+static int64 nBtfRegisterFailed = 0;
+static int64 nBtfPairOk = 0;
+static int64 nBtfPairFailed = 0;
+static int64 nBtfHandshakeNoRecv = 0;
+static int64 nBtfHandshakeNoSend = 0;
+static int64 nBtfHandshakeSilent = 0;
+static string strBtfLastDialFail;
+static string strBtfLastHandshakeTimeout;
+
+void BtfChurnNoteResolveAttempt()
+{
+    CRITICAL_BLOCK(cs_btfChurn)
+        nBtfResolveAttempts++;
+}
+
+void BtfChurnNoteResolveResult(bool fOk)
+{
+    CRITICAL_BLOCK(cs_btfChurn)
+    {
+        if (fOk) nBtfResolveOk++;
+        else    nBtfResolveMiss++;
+    }
+}
+
+void BtfChurnNoteDialAttempt(const string& strBtfAddr, const string& strMeeting)
+{
+    CRITICAL_BLOCK(cs_btfChurn)
+        nBtfDialAttempts++;
+    LogPrint("net", "btfchurn: dial attempt addr=%s meeting=%s\n",
+             strBtfAddr.c_str(), strMeeting.c_str());
+}
+
+void BtfChurnNoteDialResult(const string& strBtfAddr, const string& strMeeting, bool fOk)
+{
+    CRITICAL_BLOCK(cs_btfChurn)
+    {
+        if (fOk)
+            nBtfDialOk++;
+        else
+        {
+            nBtfDialFailed++;
+            strBtfLastDialFail = strBtfAddr + " via " + strMeeting;
+        }
+    }
+    LogPrint("net", "btfchurn: dial %s addr=%s meeting=%s\n",
+             fOk ? "ok" : "failed", strBtfAddr.c_str(), strMeeting.c_str());
+}
+
+void BtfChurnNoteRegisterResult(const string& strMeeting, bool fOk)
+{
+    CRITICAL_BLOCK(cs_btfChurn)
+    {
+        if (fOk) nBtfRegisterOk++;
+        else    nBtfRegisterFailed++;
+    }
+    LogPrint("net", "btfchurn: register %s meeting=%s\n",
+             fOk ? "ok" : "failed", strMeeting.c_str());
+}
+
+void BtfChurnNotePairResult(const string& strMeeting, bool fOk)
+{
+    CRITICAL_BLOCK(cs_btfChurn)
+    {
+        if (fOk) nBtfPairOk++;
+        else    nBtfPairFailed++;
+    }
+    LogPrint("net", "btfchurn: pair %s meeting=%s\n",
+             fOk ? "ok" : "timeout-or-drop", strMeeting.c_str());
+}
+
+void BtfChurnNoteHandshakeTimeout(const string& strBtfAddr, const string& strMeeting,
+                                  bool fRecv, bool fSend)
+{
+    CRITICAL_BLOCK(cs_btfChurn)
+    {
+        if (!fRecv && !fSend)
+            nBtfHandshakeSilent++;
+        else
+        {
+            if (!fRecv) nBtfHandshakeNoRecv++;
+            if (!fSend) nBtfHandshakeNoSend++;
+        }
+        strBtfLastHandshakeTimeout = strprintf("%s via %s recv=%d send=%d",
+                                               strBtfAddr.empty() ? "unknown" : strBtfAddr.c_str(),
+                                               strMeeting.empty() ? "unknown" : strMeeting.c_str(),
+                                               fRecv, fSend);
+    }
+}
+
 // --- Socket accounting (implementation lives in sockcount.h) --------------
 static const char* pszSockSite[SOCK_SITES] = {
     "external-ip probe", "listen socket", "inbound accept",
@@ -199,19 +296,64 @@ string GetDiagnosticsText()
 
     str += SockAccountingText();
 
-    str += "\n  peer                          dir  height   last recv   last send   unsent\n";
+    int64 nResolveAttempts, nResolveOk, nResolveMiss;
+    int64 nDialAttempts, nDialOk, nDialFailed;
+    int64 nRegisterOk, nRegisterFailed, nPairOk, nPairFailed;
+    int64 nHandshakeNoRecv, nHandshakeNoSend, nHandshakeSilent;
+    string strLastDialFail, strLastHandshakeTimeout;
+    CRITICAL_BLOCK(cs_btfChurn)
+    {
+        nResolveAttempts = nBtfResolveAttempts;
+        nResolveOk = nBtfResolveOk;
+        nResolveMiss = nBtfResolveMiss;
+        nDialAttempts = nBtfDialAttempts;
+        nDialOk = nBtfDialOk;
+        nDialFailed = nBtfDialFailed;
+        nRegisterOk = nBtfRegisterOk;
+        nRegisterFailed = nBtfRegisterFailed;
+        nPairOk = nBtfPairOk;
+        nPairFailed = nBtfPairFailed;
+        nHandshakeNoRecv = nBtfHandshakeNoRecv;
+        nHandshakeNoSend = nBtfHandshakeNoSend;
+        nHandshakeSilent = nBtfHandshakeSilent;
+        strLastDialFail = strBtfLastDialFail;
+        strLastHandshakeTimeout = strBtfLastHandshakeTimeout;
+    }
+    str += "\n  .btf churn\n";
+    str += strprintf("  resolves          %lld attempts, %lld ok, %lld no descriptor\n",
+                     (long long)nResolveAttempts, (long long)nResolveOk,
+                     (long long)nResolveMiss);
+    str += strprintf("  outbound dials    %lld attempts, %lld ok, %lld failed\n",
+                     (long long)nDialAttempts, (long long)nDialOk,
+                     (long long)nDialFailed);
+    str += strprintf("  rendezvous local  register ok/fail %lld/%lld, pair ok/drop %lld/%lld\n",
+                     (long long)nRegisterOk, (long long)nRegisterFailed,
+                     (long long)nPairOk, (long long)nPairFailed);
+    str += strprintf("  handshake grace   silent %lld, no recv %lld, no send %lld\n",
+                     (long long)nHandshakeSilent, (long long)nHandshakeNoRecv,
+                     (long long)nHandshakeNoSend);
+    if (!strLastDialFail.empty())
+        str += strprintf("  last dial failure %s\n", strLastDialFail.c_str());
+    if (!strLastHandshakeTimeout.empty())
+        str += strprintf("  last handshake    %s\n", strLastHandshakeTimeout.c_str());
+
+    str += "\n  peer                          dir  height   last recv   last send   unsent  via\n";
     foreach(CNode* pnode, vCopy)
     {
         int nSendSize = 0;
         TRY_CRITICAL_BLOCK(pnode->cs_vSend)
             nSendSize = (int)pnode->vSend.size();
-        str += strprintf("  %-28s %-4s %6d  %10s  %10s  %7d\n",
+        string strVia;
+        if (!pnode->strBtfMeeting.empty())
+            strVia = pnode->strBtfMeeting;
+        str += strprintf("  %-28s %-4s %6d  %10s  %10s  %7d  %s\n",
                          pnode->addr.ToString().substr(0, 28).c_str(),
                          pnode->fInbound ? "in" : "out",
                          pnode->nStartingHeight,
                          FormatAge(pnode->nLastRecv ? nNow - pnode->nLastRecv : -1).c_str(),
                          FormatAge(pnode->nLastSend ? nNow - pnode->nLastSend : -1).c_str(),
-                         nSendSize);
+                         nSendSize,
+                         strVia.substr(0, 28).c_str());
     }
     return str;
 }
@@ -814,30 +956,41 @@ static CNode* ConnectNodeBtfTail(const string& strBtfAddr, const unsigned char p
         return pnode;
     }
 
+    BtfChurnNoteDialAttempt(strBtfAddr, strMeeting);
     size_t colon = strMeeting.rfind(':');
     if (colon == string::npos)
+    {
+        BtfChurnNoteDialResult(strBtfAddr, strMeeting, false);
         return NULL;
+    }
     string strHost = strMeeting.substr(0, colon);
     int nPort = atoi(strMeeting.substr(colon + 1).c_str());
     if (nPort <= 0 || nPort > 65535)
+    {
+        BtfChurnNoteDialResult(strBtfAddr, strMeeting, false);
         return NULL;
+    }
 
     btf_socket_t hSocket = btf::BtfClientTunnel(strHost.c_str(), (unsigned short)nPort, pk, enc_pub);
     if (hSocket == INVALID_SOCKET)
     {
         if (fDebug)
             LogPrint("net", "ConnectNodeBtf: tunnel to %s via %s failed\n", strBtfAddr.c_str(), strMeeting.c_str());
+        BtfChurnNoteDialResult(strBtfAddr, strMeeting, false);
         return NULL;
     }
 
     if (fDebug)
         LogPrint("net", "connected %s via rendezvous %s\n", strBtfAddr.c_str(), strMeeting.c_str());
+    BtfChurnNoteDialResult(strBtfAddr, strMeeting, true);
 
     // This one answered -- worth trying first next time we start.
     RememberBtfPeer(strBtfAddr, strMeeting, enc_pub);
 
     // Add node
     pnode = new CNode(hSocket, addr, false);
+    pnode->strBtfAddr = strBtfAddr;
+    pnode->strBtfMeeting = strMeeting;
     pnode->AddRef();
     CRITICAL_BLOCK(cs_vNodes)
         vNodes.push_back(pnode);
@@ -863,12 +1016,15 @@ CNode* ConnectNodeBtf(const string& strBtfAddr)
 
     string strMeeting;
     unsigned char enc_pub[32];
+    BtfChurnNoteResolveAttempt();
     if (!BtfResolve(strBtfAddr, strMeeting, enc_pub))
     {
+        BtfChurnNoteResolveResult(false);
         if (fDebug)
             LogPrint("net", "ConnectNodeBtf: could not resolve a descriptor for %s\n", strBtfAddr.c_str());
         return NULL;
     }
+    BtfChurnNoteResolveResult(true);
     return ConnectNodeBtfTail(strBtfAddr, pk, strMeeting, enc_pub);
 }
 
@@ -951,12 +1107,14 @@ void ThreadBtfAccept(void* parg)
         }
         if (rv == btf::RV_INVALID)
         {
+            BtfChurnNoteRegisterResult(strMeeting, false);
             LogPrint("net", "rendezvous: could not register at %s, trying another\n",
                      strMeeting.c_str());
             iRelay++;       // this relay is down/attacked -> try the next one
             Sleep(3000);
             continue;
         }
+        BtfChurnNoteRegisterResult(strMeeting, true);
 
         // Registered. Advertise THIS relay now, while we are listed and before
         // anybody dials -- a descriptor naming it is what makes a dial possible
@@ -971,7 +1129,9 @@ void ThreadBtfAccept(void* parg)
         // here, the node stopped being reachable, and nothing in the log said
         // so. Re-registering every few minutes when nobody has dialled costs
         // one reconnect and removes the whole failure mode.
-        if (!btf::RvServiceWaitPaired(rv, BTF_RENDEZVOUS_WAIT_SECS))
+        bool fPaired = btf::RvServiceWaitPaired(rv, BTF_RENDEZVOUS_WAIT_SECS);
+        BtfChurnNotePairResult(strMeeting, fPaired);
+        if (!fPaired)
         {
             btf::RvClose(rv);
             LogPrint("net", "rendezvous: no dial at %s within %ds (or it dropped us), "
@@ -981,7 +1141,11 @@ void ThreadBtfAccept(void* parg)
 
         btf_socket_t hSocket = btf::BtfServiceWrap(rv, sk);
         if (hSocket == INVALID_SOCKET)
+        {
+            LogPrint("net", "rendezvous: paired at %s but channel setup failed\n",
+                     strMeeting.c_str());
             continue;
+        }
 
         // The dialer is anonymous (its pubkey never reaches us), so tag the
         // connection with a random marker address.
@@ -991,6 +1155,8 @@ void ThreadBtfAccept(void* parg)
 
         LogPrint("net", "accepted .btf connection via rendezvous %s\n", strMeeting.c_str());
         CNode* pnode = new CNode(hSocket, addr, true);
+        pnode->strBtfAddr = "inbound";
+        pnode->strBtfMeeting = strMeeting;
         pnode->AddRef();
         CRITICAL_BLOCK(cs_vNodes)
             vNodes.push_back(pnode);
@@ -1457,6 +1623,9 @@ void ThreadSocketHandler2(void* parg)
             {
                 if (pnode->nLastRecv == 0 || pnode->nLastSend == 0)
                 {
+                    if (!pnode->strBtfMeeting.empty())
+                        BtfChurnNoteHandshakeTimeout(pnode->strBtfAddr, pnode->strBtfMeeting,
+                                                     pnode->nLastRecv != 0, pnode->nLastSend != 0);
                     LogPrint("net", "socket no message in first %d seconds, recv=%d send=%d\n",
                              BTF_HANDSHAKE_GRACE_SECS, pnode->nLastRecv != 0, pnode->nLastSend != 0);
                     pnode->fDisconnect = true;
@@ -1803,9 +1972,22 @@ bool StartNode(string& strError)
 
     // Anonymous inbound: this node's .btf hidden service, reachable through the
     // meeting relay without exposing our IP or needing a public port.
+    //
+    // Several of these, not one. Each holds its own registration at a
+    // rendezvous, and a registration is consumed the moment a caller is paired
+    // with it -- so with a single thread the node is unreachable for the whole
+    // gap between being paired and registering again. Measured from a node's
+    // own counters: half of all outbound dials failed for exactly that reason,
+    // 15 of 30 in eleven minutes, while descriptors resolved 53 times out of
+    // 54. Spares parked at the relay close that gap.
+    //
+    // Needs a relay that keeps more than one registration per node; against an
+    // older relay the extra threads are harmless, because it drops the previous
+    // registration on each new one and the node ends up where it started.
     if (!vBtfMeetingRelays.empty())
-        if (_beginthread(ThreadBtfAccept, 0, NULL) == -1)
-            printf("Error: _beginthread(ThreadBtfAccept) failed\n");
+        for (int i = 0; i < BTF_ACCEPT_THREADS; i++)
+            if (_beginthread(ThreadBtfAccept, 0, NULL) == -1)
+                printf("Error: _beginthread(ThreadBtfAccept) failed\n");
 
     // Anonymous outbound: keep a connection to a specific .btf peer, if asked.
     if (!strBtfConnect.empty())
diff --git a/src/net.h b/src/net.h
index 37f335e..59a8028 100644
--- a/src/net.h
+++ b/src/net.h
@@ -54,6 +54,12 @@ static const int64        PEX_MIN_INTERVAL         = 60;
 // traffic is nothing: one per relay per five minutes.
 static const int          BTF_RENDEZVOUS_WAIT_SECS = 300;
 
+// Registrations this node parks at a rendezvous at once. One is what the
+// protocol needs; more is what keeps the node reachable while one of them is
+// being paired, since pairing consumes the registration it used. Four matches
+// the relay's ceiling per node.
+static const int          BTF_ACCEPT_THREADS       = 3;
+
 // Liveness of an established peer connection. BTF_RENDEZVOUS_WAIT_SECS bounds
 // the wait for a dial to arrive; these bound the connection that comes out of
 // it, which has the same failure mode once it goes quiet.
@@ -139,6 +145,18 @@ extern int nBestHeight;
 // absurd height cannot move it.
 int GetPeerMedianHeight();
 
+// Lightweight .btf churn accounting. These counters are diagnostics only: they
+// let an operator tell stale descriptors, relay pairing failures, and paired
+// tunnels that never speak apart before changing behaviour.
+void BtfChurnNoteResolveAttempt();
+void BtfChurnNoteResolveResult(bool fOk);
+void BtfChurnNoteDialAttempt(const string& strBtfAddr, const string& strMeeting);
+void BtfChurnNoteDialResult(const string& strBtfAddr, const string& strMeeting, bool fOk);
+void BtfChurnNoteRegisterResult(const string& strMeeting, bool fOk);
+void BtfChurnNotePairResult(const string& strMeeting, bool fOk);
+void BtfChurnNoteHandshakeTimeout(const string& strBtfAddr, const string& strMeeting,
+                                  bool fRecv, bool fSend);
+
 // Descriptors to hand a peer: ours first, then peers that answered us.
 void BtfPexCollect(std::vector<std::string>& vDescOut);
 // Verify descriptors a peer sent and remember the good ones. Returns how many.
@@ -606,6 +624,8 @@ public:
     int64 nIncompleteMessageStart;
     unsigned int nIncompleteMessageSize;
     string strIncompleteMessageCommand;
+    string strBtfAddr;
+    string strBtfMeeting;
 
     // Height this peer announced in its version message, or -1 if it sent a
     // version message that predates the field. Knowing how far along everyone
@@ -639,6 +659,8 @@ public:
         nIncompleteMessageStart = 0;
         nIncompleteMessageSize = 0;
         strIncompleteMessageCommand.clear();
+        strBtfAddr.clear();
+        strBtfMeeting.clear();
         nStartingHeight = -1;
         vfSubscribe.assign(256, false);
 
diff --git a/src/nostr.cpp b/src/nostr.cpp
index dc96be2..1fd60ad 100644
--- a/src/nostr.cpp
+++ b/src/nostr.cpp
@@ -1596,6 +1596,8 @@ static void ConnectDiscoveredBtfPeers()
     // handshakes refused, which then looks exactly like "dead peers" even
     // though the descriptors were perfectly resolvable.
     map<string, BtfResolvedPeer> resolved;
+    for (const string& addr : candidates)
+        BtfChurnNoteResolveAttempt();
     BtfResolveMany(candidates, resolved);
 
     // Candidates whose descriptor didn't resolve this pass are recorded as a
@@ -1608,9 +1610,11 @@ static void ConnectDiscoveredBtfPeers()
         {
             if (resolved.count(addr))
             {
+                BtfChurnNoteResolveResult(true);
                 toDial.push_back(addr);
                 continue;
             }
+            BtfChurnNoteResolveResult(false);
             int n = ++g_btfPeerFails[addr];
             int64 cap = (n >= BTF_CHRONIC_FAIL_THRESHOLD) ? BTF_BACKOFF_CAP_SECS_DEAD
                                                            : BTF_BACKOFF_CAP_SECS;