Close REALITY relay cleanly when event monitoring fails

src/transport_layer/reality_relay.c

@@ -131,7 +131,12 @@ static void relay_flush_c2d(struct reality_relay *r) {
         if (sent < 0) {
             int err = socket_get_error();
             if (err == ERR_INTR) continue;
-            if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) { uasync_set_socket_write(r->ua, r->dest_sid, 1); return; }
+            if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) {
+                if (uasync_set_socket_write(r->ua, r->dest_sid, 1) != ERR_OK) {
+                    DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot wait for dest write"); relay_free(r);
+                }
+                return;
+            }
             DEBUG_WARN(DEBUG_CATEGORY_REALITY, "reality_relay send to dest failed err=%d", err);
             relay_free(r); return;
         }
@@ -141,8 +146,10 @@ static void relay_flush_c2d(struct reality_relay *r) {
         if (relay_touch(r) < 0) { relay_free(r); return; }
     }
     u_free(r->c2d); r->c2d = NULL;
-    uasync_set_socket_write(r->ua, r->dest_sid, 0);
-    if (!r->client_read_eof) uasync_set_socket_read(r->ua, r->client_sid, 1);
+    if (uasync_set_socket_write(r->ua, r->dest_sid, 0) != ERR_OK ||
+        (!r->client_read_eof && uasync_set_socket_read(r->ua, r->client_sid, 1) != ERR_OK)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot resume client after dest flush"); relay_free(r);
+    }
 }
 
 static void relay_flush_d2c(struct reality_relay *r) {
@@ -152,7 +159,12 @@ static void relay_flush_d2c(struct reality_relay *r) {
         if (sent < 0) {
             int err = socket_get_error();
             if (err == ERR_INTR) continue;
-            if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) { uasync_set_socket_write(r->ua, r->client_sid, 1); return; }
+            if (err == ERR_AGAIN || err == ERR_WOULDBLOCK) {
+                if (uasync_set_socket_write(r->ua, r->client_sid, 1) != ERR_OK) {
+                    DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot wait for client write"); relay_free(r);
+                }
+                return;
+            }
             DEBUG_WARN(DEBUG_CATEGORY_REALITY, "reality_relay send to client failed err=%d", err);
             relay_free(r); return;
         }
@@ -162,14 +174,16 @@ static void relay_flush_d2c(struct reality_relay *r) {
         if (relay_touch(r) < 0) { relay_free(r); return; }
     }
     u_free(r->d2c); r->d2c = NULL;
-    uasync_set_socket_write(r->ua, r->client_sid, 0);
-    if (!r->dest_read_eof) uasync_set_socket_read(r->ua, r->dest_sid, 1);
+    if (uasync_set_socket_write(r->ua, r->client_sid, 0) != ERR_OK ||
+        (!r->dest_read_eof && uasync_set_socket_read(r->ua, r->dest_sid, 1) != ERR_OK)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot resume dest after client flush"); relay_free(r);
+    }
 }
 
 static void relay_client_read_cb(socket_t sock, void *arg) {
     struct reality_relay *r = (struct reality_relay *)arg;
     (void)sock;
-    if (r->c2d) return;
+    if (r->closed || r->c2d) return;
     uint8_t *buf = u_malloc(RELAY_IO_BUF);
     if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay client receive allocation failed"); relay_free(r); return; }
     ssize_t n = recv(r->client_sock, buf, RELAY_IO_BUF, 0);
@@ -183,7 +197,9 @@ static void relay_client_read_cb(socket_t sock, void *arg) {
         u_free(buf);
         r->client_read_eof = 1;
         DEBUG_DEBUG(DEBUG_CATEGORY_REALITY, "relay client EOF: dest_eof=%d", r->dest_read_eof);
-        uasync_set_socket_read(r->ua, r->client_sid, 0);
+        if (uasync_set_socket_read(r->ua, r->client_sid, 0) != ERR_OK) {
+            DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot stop client read at EOF"); relay_free(r); return;
+        }
         if (shutdown(r->dest_sock, RELAY_SHUT_WR) < 0)
             DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay dest half-close failed: err=%d", socket_get_error());
         if (r->dest_read_eof) relay_free(r);
@@ -200,7 +216,9 @@ static void relay_client_read_cb(socket_t sock, void *arg) {
     }
 #endif
     r->c2d = buf; r->c2d_len = (size_t)n; r->c2d_off = 0;
-    uasync_set_socket_read(r->ua, r->client_sid, 0);   // backpressure
+    if (uasync_set_socket_read(r->ua, r->client_sid, 0) != ERR_OK) {
+        DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot pause client for backpressure"); relay_free(r); return;
+    }
     if (relay_touch(r) < 0) { relay_free(r); return; }
     relay_flush_c2d(r);
 }
@@ -208,7 +226,7 @@ static void relay_client_read_cb(socket_t sock, void *arg) {
 static void relay_dest_read_cb(socket_t sock, void *arg) {
     struct reality_relay *r = (struct reality_relay *)arg;
     (void)sock;
-    if (r->d2c) return;
+    if (r->closed || r->d2c) return;
     uint8_t *buf = u_malloc(RELAY_IO_BUF);
     if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay dest receive allocation failed"); relay_free(r); return; }
     ssize_t n = recv(r->dest_sock, buf, RELAY_IO_BUF, 0);
@@ -222,14 +240,18 @@ static void relay_dest_read_cb(socket_t sock, void *arg) {
         u_free(buf);
         r->dest_read_eof = 1;
         DEBUG_DEBUG(DEBUG_CATEGORY_REALITY, "relay dest EOF: client_eof=%d", r->client_read_eof);
-        uasync_set_socket_read(r->ua, r->dest_sid, 0);
+        if (uasync_set_socket_read(r->ua, r->dest_sid, 0) != ERR_OK) {
+            DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot stop dest read at EOF"); relay_free(r); return;
+        }
         if (shutdown(r->client_sock, RELAY_SHUT_WR) < 0)
             DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay client half-close failed: err=%d", socket_get_error());
         if (r->client_read_eof) relay_free(r);
         return;
     }
     r->d2c = buf; r->d2c_len = (size_t)n; r->d2c_off = 0;
-    uasync_set_socket_read(r->ua, r->dest_sid, 0);      // backpressure
+    if (uasync_set_socket_read(r->ua, r->dest_sid, 0) != ERR_OK) {
+        DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot pause dest for backpressure"); relay_free(r); return;
+    }
     if (relay_touch(r) < 0) { relay_free(r); return; }
     relay_flush_d2c(r);
 }
@@ -237,12 +259,19 @@ static void relay_dest_read_cb(socket_t sock, void *arg) {
 static void relay_client_write_cb(socket_t sock, void *arg) {
     struct reality_relay *r = (struct reality_relay *)arg;
     (void)sock;
-    if (!r->d2c) { uasync_set_socket_write(r->ua, r->client_sid, 0); return; }
+    if (r->closed) return;
+    if (!r->d2c) {
+        if (uasync_set_socket_write(r->ua, r->client_sid, 0) != ERR_OK) {
+            DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot stop idle client write"); relay_free(r);
+        }
+        return;
+    }
     relay_flush_d2c(r);
 }
 
 static void relay_dest_write_cb(socket_t sock, void *arg) {
     struct reality_relay *r = (struct reality_relay *)arg;
+    if (r->closed) return;
     if (r->connecting) {
         int err = 0; socklen_t elen = sizeof(err);
         if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &elen) < 0) err = socket_get_error();
@@ -255,10 +284,18 @@ static void relay_dest_write_cb(socket_t sock, void *arg) {
         if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&opt, sizeof(opt)) < 0)
             DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay dest TCP_NODELAY failed: err=%d", socket_get_error());
         if (r->c2d) relay_flush_c2d(r);
-        else { uasync_set_socket_write(r->ua, r->dest_sid, 0); uasync_set_socket_read(r->ua, r->client_sid, 1); }
+        else if (uasync_set_socket_write(r->ua, r->dest_sid, 0) != ERR_OK ||
+                 uasync_set_socket_read(r->ua, r->client_sid, 1) != ERR_OK) {
+            DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot enable connected stream"); relay_free(r);
+        }
+        return;
+    }
+    if (!r->c2d) {
+        if (uasync_set_socket_write(r->ua, r->dest_sid, 0) != ERR_OK) {
+            DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot stop idle dest write"); relay_free(r);
+        }
         return;
     }
-    if (!r->c2d) { uasync_set_socket_write(r->ua, r->dest_sid, 0); return; }
     relay_flush_c2d(r);
 }
 
@@ -274,6 +311,7 @@ static void relay_except_cb(socket_t sock, void *arg) {
         relay_free(r); return;
     }
     if (sock == r->client_sock) {
+        if (r->connecting) { DEBUG_DEBUG(DEBUG_CATEGORY_REALITY, "relay client closed before dest connection"); relay_free(r); return; }
         if (!r->client_read_eof) relay_client_read_cb(sock, r);
     } else if (!r->connecting && !r->dest_read_eof) relay_dest_read_cb(sock, r);
 }
@@ -312,7 +350,10 @@ static int relay_connect_dest(struct reality_relay *r, const struct sockaddr_in
 
     if (!r->connecting) {
         if (r->c2d) relay_flush_c2d(r);
-        else uasync_set_socket_read(r->ua, r->client_sid, 1);
+        else if (uasync_set_socket_write(r->ua, r->dest_sid, 0) != ERR_OK ||
+                 uasync_set_socket_read(r->ua, r->client_sid, 1) != ERR_OK) {
+            DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot enable immediately connected stream"); relay_free(r);
+        }
     }
     return r->closed ? -1 : 0;
 }
@@ -364,7 +405,9 @@ int reality_relay_start(struct reality_owner *owner, struct UASYNC *ua, socket_t
     r->client_sid = uasync_add_socket_t(ua, client_sock, relay_client_read_cb, relay_client_write_cb, relay_except_cb, "reality_client", r);
     if (!r->client_sid) { DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay client registration failed"); relay_free(r); return -1; }
     // не читаем от клиента, пока dest не подключён (иначе буферили бы без границ во время DNS)
-    uasync_set_socket_read(ua, r->client_sid, 0);
+    if (uasync_set_socket_read(ua, r->client_sid, 0) != ERR_OK || uasync_set_socket_write(ua, r->client_sid, 0) != ERR_OK) {
+        DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay cannot suspend client during connect"); relay_free(r); return -1;
+    }
 
     // парсим "host:port"
     {

tests/Makefile.am

@@ -271,7 +271,8 @@ check_PROGRAMS += test_socket_monitor
 check_PROGRAMS += test_reality_relay_faults
 test_reality_relay_faults_SOURCES = test_reality_relay_faults.c
 test_reality_relay_faults_LDADD = $(LIBUTUN) $(COMMON_LIBS)
-test_reality_relay_faults_LDFLAGS = -Wl,--wrap=uasync_set_timeout -Wl,--wrap=uasync_call_soon -Wl,--wrap=socket_set_nonblocking
+test_reality_relay_faults_LDFLAGS = -Wl,--wrap=uasync_set_timeout -Wl,--wrap=uasync_call_soon \
+    -Wl,--wrap=socket_set_nonblocking -Wl,--wrap=uasync_set_socket_read
 check_PROGRAMS += test_call_ring
 test_call_ring_SOURCES = test_call_ring.c ../src/transport_layer/call_ring.c
 test_call_ring_LDADD = $(COMMON_LIBS)

tests/test_reality_relay_faults.c

@@ -11,7 +11,12 @@
 #include <limits.h>
 #include <string.h>
 
-static int fail_idle, fail_free, fail_nonblock;
+static int fail_idle, fail_free, fail_nonblock, fail_monitor;
+extern err_t __real_uasync_set_socket_read(struct UASYNC*, void*, int);
+err_t __wrap_uasync_set_socket_read(struct UASYNC* ua, void* sid, int enabled) {
+    if (fail_monitor) { fail_monitor = 0; return ERR_FAIL; }
+    return __real_uasync_set_socket_read(ua, sid, enabled);
+}
 extern void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*);
 void* __wrap_uasync_set_timeout(struct UASYNC* ua, int delay, void* arg, timeout_callback_t cb, const char* name) {
     if (fail_idle && !strcmp(name, "relay_idle") && !--fail_idle) return NULL;
@@ -57,9 +62,13 @@ static void test_failures(void) {
     client_pair(pair); fail_nonblock = 1;
     assert(reality_relay_start(&owner, ua, pair[0], "127.0.0.1:1", NULL, 0, 1) == -1);
     socket_close_wrapper(pair[1]); reality_owner_cleanup(&owner);
+    client_pair(pair); fail_monitor = 1;
+    assert(reality_relay_start(&owner, ua, pair[0], "127.0.0.1:1", NULL, 0, 1) == -1);
+    assert(!owner.relay_count && !owner.relays);
+    socket_close_wrapper(pair[1]); reality_owner_cleanup(&owner);
     uasync_drain_immediate(ua); uasync_destroy(ua, 1);
     assert(u_get_allocated_count() == initial);
-    DEBUG_INFO(DEBUG_CATEGORY_REALITY, "PASS relay invalid ports, timeout range, timer/nonblocking failures and retained free context cleanup");
+    DEBUG_INFO(DEBUG_CATEGORY_REALITY, "PASS relay invalid ports, timeout range, timer/nonblocking/monitor failures and retained free cleanup");
 }
 
 static void test_stream(void) {