diff --git a/src/transport_layer/reality_relay.c b/src/transport_layer/reality_relay.c index ae9e8b65..bf11588e 100644 --- a/src/transport_layer/reality_relay.c +++ b/src/transport_layer/reality_relay.c @@ -72,6 +72,7 @@ static void relay_client_read_cb(socket_t sock, void *arg); static void relay_dest_read_cb(socket_t sock, void *arg); static void relay_client_write_cb(socket_t sock, void *arg); static void relay_dest_write_cb(socket_t sock, void *arg); +static void relay_except_cb(socket_t sock, void *arg); static void relay_free_cb(void *arg) { struct reality_relay* r = arg; @@ -105,7 +106,7 @@ static void relay_free(struct reality_relay *r) { if (!r->free_call) DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay free scheduling failed; context retained until owner cleanup"); } -// Перезапустить таймер простоя: сброс на любую активность (чтение данных). +// Перезапускает таймер простоя после чтения или успешной отправки данных. static int relay_touch(struct reality_relay *r) { if (r->idle_timeout_tb <= 0) return 0; void* timer = uasync_set_timeout(r->ua, r->idle_timeout_tb, r, relay_idle_timeout_cb, "relay_idle"); @@ -136,9 +137,9 @@ static void relay_flush_c2d(struct reality_relay *r) { } if (sent == 0) { DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay send to dest made no progress"); relay_free(r); return; } r->c2d_off += (size_t)sent; + r->bytes_c2d += (size_t)sent; if (relay_touch(r) < 0) { relay_free(r); return; } } - r->bytes_c2d += r->c2d_len; 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); @@ -157,9 +158,9 @@ static void relay_flush_d2c(struct reality_relay *r) { } if (sent == 0) { DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay send to client made no progress"); relay_free(r); return; } r->d2c_off += (size_t)sent; + r->bytes_d2c += (size_t)sent; if (relay_touch(r) < 0) { relay_free(r); return; } } - r->bytes_d2c += r->d2c_len; 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); @@ -170,7 +171,7 @@ static void relay_client_read_cb(socket_t sock, void *arg) { (void)sock; if (r->c2d) return; uint8_t *buf = u_malloc(RELAY_IO_BUF); - if (!buf) { relay_free(r); return; } + 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); if (n < 0) { int err = socket_get_error(); @@ -181,8 +182,10 @@ static void relay_client_read_cb(socket_t sock, void *arg) { if (n == 0) { 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); - shutdown(r->dest_sock, RELAY_SHUT_WR); + 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); return; } @@ -207,7 +210,7 @@ static void relay_dest_read_cb(socket_t sock, void *arg) { (void)sock; if (r->d2c) return; uint8_t *buf = u_malloc(RELAY_IO_BUF); - if (!buf) { relay_free(r); return; } + 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); if (n < 0) { int err = socket_get_error(); @@ -218,8 +221,10 @@ static void relay_dest_read_cb(socket_t sock, void *arg) { if (n == 0) { 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); - shutdown(r->client_sock, RELAY_SHUT_WR); + 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; } @@ -240,12 +245,15 @@ static void relay_dest_write_cb(socket_t sock, void *arg) { struct reality_relay *r = (struct reality_relay *)arg; if (r->connecting) { int err = 0; socklen_t elen = sizeof(err); - if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &elen) < 0 || err != 0) { + if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *)&err, &elen) < 0) err = socket_get_error(); + if (err != 0) { DEBUG_WARN(DEBUG_CATEGORY_REALITY, "reality_relay connect to dest failed err=%d", err); relay_free(r); return; } r->connecting = 0; - int opt = 1; setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&opt, sizeof(opt)); + int opt = 1; + 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); } return; @@ -254,6 +262,22 @@ static void relay_dest_write_cb(socket_t sock, void *arg) { relay_flush_c2d(r); } +/* HUP может прийти с последними данными: сначала дочитать поток, сохранив backpressure. */ +static void relay_except_cb(socket_t sock, void *arg) { + struct reality_relay* r = arg; + if (r->closed) return; + int err = 0; socklen_t size = sizeof(err); + if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char*)&err, &size) < 0) err = socket_get_error(); + if (err) { + DEBUG_WARN(DEBUG_CATEGORY_REALITY, "relay socket exception: side=%s err=%d", + sock == r->client_sock ? "client" : "dest", err); + relay_free(r); return; + } + if (sock == r->client_sock) { + 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); +} + // Создать неблокирующий dest-сокет и начать connect() к уже разрезолвенному адресу. // Вызывается сразу (literal IP) или из коллбэка async DNS. static int relay_connect_dest(struct reality_relay *r, const struct sockaddr_in *addr) { @@ -278,10 +302,12 @@ static int relay_connect_dest(struct reality_relay *r, const struct sockaddr_in } } else { r->connecting = 0; - int opt = 1; setsockopt(dest_sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&opt, sizeof(opt)); + int opt = 1; + if (setsockopt(dest_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()); } - r->dest_sid = uasync_add_socket_t(r->ua, dest_sock, relay_dest_read_cb, relay_dest_write_cb, NULL, "reality_dest", r); + r->dest_sid = uasync_add_socket_t(r->ua, dest_sock, relay_dest_read_cb, relay_dest_write_cb, relay_except_cb, "reality_dest", r); if (!r->dest_sid) { DEBUG_ERROR(DEBUG_CATEGORY_REALITY, "relay dest registration failed"); relay_free(r); return -1; } if (!r->connecting) { @@ -335,7 +361,7 @@ int reality_relay_start(struct reality_owner *owner, struct UASYNC *ua, socket_t // берём владение client_sock сразу (снимаем старую регистрацию) — до парсинга, // чтобы любая ошибка ниже не оставила сокет зарегистрированным в uasync. - r->client_sid = uasync_add_socket_t(ua, client_sock, relay_client_read_cb, relay_client_write_cb, NULL, "reality_client", r); + 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); diff --git a/tests/test_reality_relay_faults.c b/tests/test_reality_relay_faults.c index 2401ff03..9ec73ce5 100644 --- a/tests/test_reality_relay_faults.c +++ b/tests/test_reality_relay_faults.c @@ -72,7 +72,7 @@ static void test_stream(void) { assert(!socket_set_nonblocking(listener)); char destination[64]; snprintf(destination, sizeof(destination), "127.0.0.1:%u", test_bound_port(listener)); socket_t pair[2]; client_pair(pair); - assert(!reality_relay_start(&owner, ua, pair[0], destination, (const uint8_t*)"hello", 5, 1)); + assert(!reality_relay_start(&owner, ua, pair[0], destination, (const uint8_t*)"hello", 5, 0)); socket_t remote = SOCKET_INVALID; for (unsigned i = 0; remote == SOCKET_INVALID && i < 1000; i++) { remote = accept(listener, NULL, NULL); uasync_poll(ua, 10); } assert(remote != SOCKET_INVALID && !socket_set_nonblocking(remote));