rendezvous: let a node park more than one registration (#113)

A registration was consumed by the first caller and registering again replaced the previous one, so a node was reachable to one peer at a time and unreachable for the gap between being paired and registering again. Measured on a live node: descriptors resolved 53/54, our own registrations paired 11/12, nothing paired and went silent -- and half of every outbound dial failed, 15 of 30 in eleven minutes.

The relay keeps a queue per pubkey capped at four, pairs with the oldest still-alive registration and discards the dead; the node runs three accept threads, each holding its own registration. The wire protocol does not change, so either side can be deployed first.

Bench with both relays built from source and waves of concurrent dialers: old+1 65%, new+1 58%, old+3 68%, new+3 98%. Each half alone lands within noise of doing nothing.

1 files changed, 73 insertions, 24 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)