1 files changed, 69 insertions, 5 deletions

diff --git a/src/btfrv.cpp b/src/btfrv.cpp
index 905717f..7d705ef 100644
--- a/src/btfrv.cpp
+++ b/src/btfrv.cpp
@@ -5,6 +5,9 @@
 // Windows/Winsock implementation (Layer 6 will #ifdef this for Linux/BSD sockets).
 
 #include "btfrv.h"
+#ifdef BITFLASH_NODE_BUILD
+#include "proxy.h"
+#endif
 
 // Portable sockets: Winsock on Windows, BSD sockets on Linux/POSIX (for the
 // headless relay `bitflashd` that runs on the VPS).
@@ -37,6 +40,8 @@ typedef int SOCKET;
 #include <thread>
 #include <mutex>
 #include <atomic>
+#include <chrono>
+#include <system_error>
 
 namespace btf
 {
@@ -92,6 +97,11 @@ static bool WriteN(SOCKET s, const void* buf, int n)
 
 static SOCKET ConnectTo(const char* host, unsigned short port)
 {
+#ifdef BITFLASH_NODE_BUILD
+    SOCKET s = BtfConnectSocket(host, port, SOCK_RV_DIAL, 10);
+    if (s == INVALID_SOCKET)
+        return INVALID_SOCKET;
+#else
     struct addrinfo hints, *res = NULL;
     memset(&hints, 0, sizeof(hints));
     hints.ai_family = AF_UNSPEC; // IPv4 or IPv6 (some relays are v6-only)
@@ -130,6 +140,7 @@ static SOCKET ConnectTo(const char* host, unsigned short port)
     }
     freeaddrinfo(res);
     if (s == INVALID_SOCKET) return INVALID_SOCKET;
+#endif
 
     // Clear the timeout -- the socket is now a live tunnel and must not
     // time out when idle. recv() will block until data arrives or peer closes.
@@ -200,8 +211,28 @@ static void HandleIncoming(SOCKET s)
         bool a = WriteN(svc, &ok, 1);
         bool b = WriteN(s, &ok, 1);
         if (!a || !b) { CLOSESOCK(svc); CLOSESOCK(s); return; }
-        std::thread(Forward, s, svc).detach();
-        std::thread(Forward, svc, s).detach();
+        // One thread pumps s -> svc; this thread pumps svc -> s. Spawning a
+        // thread per direction left no clean way to fail the second spawn:
+        // std::thread throws std::system_error when pthread_create is refused
+        // (NPROC limit, container pids cgroup, memory pressure), the first
+        // pump was already detached, and closing its sockets from here would
+        // race it -- so nothing caught the throw and it reached
+        // std::terminate, killing the relay over one pair it could not serve.
+        // Running one direction on this thread leaves exactly one spawn to
+        // guard, and a failure here has nothing running yet to race.
+        try
+        {
+            std::thread(Forward, s, svc).detach();
+        }
+        catch (const std::system_error& e)
+        {
+            fprintf(stderr, "rendezvous relay: cannot spawn forward thread (%s), pair dropped\n",
+                    e.what());
+            CLOSESOCK(svc);
+            CLOSESOCK(s);
+            return;
+        }
+        Forward(svc, s);
     }
     else
     {
@@ -227,6 +258,7 @@ bool RvRelayRun(unsigned short port)
     if (bind(g_listen, (struct sockaddr*)&addr, sizeof(addr)) != 0) { CLOSESOCK(g_listen); return false; }
     if (listen(g_listen, 16) != 0) { CLOSESOCK(g_listen); return false; }
 
+    int nSpawnFails = 0;
     while (!g_stop)
     {
         struct sockaddr_in cli;
@@ -237,7 +269,27 @@ bool RvRelayRun(unsigned short port)
             if (g_stop) break;
             continue;
         }
-        std::thread(HandleIncoming, s).detach();
+        try
+        {
+            std::thread(HandleIncoming, s).detach();
+            nSpawnFails = 0;
+        }
+        catch (const std::system_error& e)
+        {
+            // A refused pthread_create is backpressure, not a reason to die:
+            // the condition behind it (NPROC, pids cgroup, memory) does not
+            // clear in milliseconds, and the connection count feeding it is
+            // attacker-influenced. Drop this one peer, stop accepting for a
+            // moment -- longer each time, capped -- and keep serving the
+            // connections already established.
+            fprintf(stderr, "rendezvous relay: cannot spawn connection thread (%s), connection dropped\n",
+                    e.what());
+            CLOSESOCK(s);
+            nSpawnFails++;
+            int nWaitMs = nSpawnFails < 5 ? nSpawnFails * 1000 : 5000;
+            for (int i = 0; i < nWaitMs && !g_stop; i += 100)
+                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        }
     }
     CLOSESOCK(g_listen);
     g_listen = INVALID_SOCKET;
@@ -325,17 +377,29 @@ bool RvServiceWaitPaired(RvSocket s, int nTimeoutSecs)
 }
 
 RvSocket RvClientConnect(const char* relay_host, unsigned short port,
-                         const unsigned char target_pubkey[32])
+                         const unsigned char target_pubkey[32],
+                         RvClientFail* pFail)
 {
+    if (pFail) *pFail = RVC_FAIL_DIAL;
     if (!RvInit()) return RV_INVALID;
     SOCKET s = ConnectTo(relay_host, port);
     if (s == INVALID_SOCKET) return RV_INVALID;
+    if (pFail) *pFail = RVC_FAIL_PAIR;
     unsigned char hdr[33];
     hdr[0] = (unsigned char)'C';
     memcpy(hdr + 1, target_pubkey, 32);
     if (!WriteN(s, hdr, 33)) { CLOSESOCK(s); return RV_INVALID; }
     unsigned char paired = 0;
-    if (!ReadN(s, &paired, 1) || paired != 0x01) { CLOSESOCK(s); return RV_INVALID; }
+    if (!ReadN(s, &paired, 1)) { CLOSESOCK(s); return RV_INVALID; }
+    if (paired != 0x01)
+    {
+        // The relay answered and declined: HandleIncoming writes 0x00 and
+        // closes when no service is parked under this pubkey right now.
+        if (pFail) *pFail = RVC_FAIL_REFUSED;
+        CLOSESOCK(s);
+        return RV_INVALID;
+    }
+    if (pFail) *pFail = RVC_FAIL_NONE;
     return (RvSocket)s;
 }