From 758ff4bc49af78b06e37fc1e99741cb5fa8e43d6 Mon Sep 17 00:00:00 2001 From: Bitflash Date: Sun, 2 Aug 2026 11:59:12 -0300 Subject: 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. --- relay/btfrv.cpp | 97 +++++++++++++++++++++------- src/net.cpp | 194 ++++++++++++++++++++++++++++++++++++++++++++++++++++++-- src/net.h | 22 +++++++ src/nostr.cpp | 4 ++ 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 #include #include +#include #include #include #include @@ -44,7 +45,24 @@ static std::atomic 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 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 > 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 lk(g_mtx); - nBefore = (int)g_waiting.size(); - std::map::iterator it = g_waiting.begin(); + std::map >::iterator it = g_waiting.begin(); while (it != g_waiting.end()) { - if (RegistrationDead(it->second)) + std::deque& q = it->second; + nBefore += (int)q.size(); + for (std::deque::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 lk(g_mtx); - // If a stale service for this key exists, drop it. - std::map::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& 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::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 lk(g_mtx); - std::map::iterator it = g_waiting.find(pubkey); + std::map >::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& 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& 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 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; -- cgit v1.2.3