diff --git a/src/transport_layer/etcp.h b/src/transport_layer/etcp.h index 0734f9eb..0a67870a 100644 --- a/src/transport_layer/etcp.h +++ b/src/transport_layer/etcp.h @@ -121,7 +121,6 @@ struct ETCP_CONN { // Links (channels) - linked list struct ETCP_LINK* links; struct ETCP_LINK* last_rr_link; // последний линк, выбранный round-robin - uint8_t tcp_link_count; // количество UP TCP линков uint8_t last_assigned_link_id; // последний выданный link_id (для циклического перебора) struct ETCP_PADDING padding; // диапазон и независимый таймер всех линков этого соединения diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c index adb16ada..30ed932e 100644 --- a/src/transport_layer/etcp_connections.c +++ b/src/transport_layer/etcp_connections.c @@ -65,6 +65,7 @@ static int socket_identity_init(struct ETCP_SOCKET *socket, struct UTUN_INSTANCE * Только static-функции, используемые раньше своего определения. * Не-static функции уже объявлены в etcp_connections.h. */ static void tcp_link_close_cb(struct stcp_link *sl, int err, void *arg); +static int etcp_link_rx_alive(struct ETCP_CONN* conn, struct ETCP_LINK* link); /* Wire fields may start at any byte offset. */ static uint16_t wire_read16(const void* p) { uint16_t v; memcpy(&v,p,2); return be16toh(v); } static uint64_t wire_read64(const void* p) { uint64_t v; memcpy(&v,p,8); return be64toh(v); } @@ -1188,7 +1189,12 @@ static void etcp_inflight_nullify_link(struct ll_queue* q, struct ETCP_LINK* dea // зануление ссылок inflight (защита от use-after-free) и освобождение памяти. void etcp_link_close(struct ETCP_LINK* link) { DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - if (!link) return; + if (!link || link->closing) return; + link->closing = 1; + struct ETCP_CONN *etcp = link->etcp; + etcp->callbacks_running++; + int old_state = link->link_state, old_status = link->link_status; + link->link_status = link->recv_keepalive = link->initialized = 0; if (link->rx_timeout_timer) { uasync_cancel_timeout(link->etcp->instance->ua, link->rx_timeout_timer); link->rx_timeout_timer = NULL; } @@ -1205,22 +1211,8 @@ void etcp_link_close(struct ETCP_LINK* link) { uasync_cancel_timeout(link->etcp->instance->ua, link->burst_resp_timer); link->burst_resp_timer = NULL; } - etcp_tx_link_destroy(link); - - if (!link->conn) { - struct ETCP_LINK **pp = &link->etcp->links; - while (*pp && *pp != link) pp = &(*pp)->next; - if (*pp) *pp = link->next; - if (link->init_timer) uasync_cancel_timeout(link->etcp->instance->ua, link->init_timer); - if (link->shaper_timer) uasync_cancel_timeout(link->etcp->instance->ua, link->shaper_timer); - if (link->keepalive_timer) uasync_cancel_timeout(link->etcp->instance->ua, link->keepalive_timer); - etcp_conn_on_inflight_lim_changed(link->etcp); - etcp_inflight_nullify_link(link->etcp->input_wait_ack, link); - etcp_inflight_nullify_link(link->etcp->input_send_q, link); - callring_put(CR_LINK_CLOSE_FREE, (uintptr_t)link, 0); - u_free(link->bbr); - u_free(link); - return; + if (link->ka_sleep_announce_timer) { + uasync_cancel_timeout(etcp->instance->ua, link->ka_sleep_announce_timer); link->ka_sleep_announce_timer = NULL; } // Cancel init timer if active @@ -1252,18 +1244,19 @@ void etcp_link_close(struct ETCP_LINK* link) { } remove_link_from_queue(link); - + etcp_tx_link_destroy(link); etcp_conn_on_inflight_lim_changed(link->etcp); DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] Link %d closed: addr=%s rcvd=%zub ack=%llub infl=%ub/%upkt", link->etcp->log_name, link->local_link_id, sockaddr_storage_to_str(&link->remote_addr).str, link->total_decrypted, (unsigned long long)link->acked_bytes, link->inflight_bytes, link->inflight_packets); - etcp_fire_link_status_cbk(link, link->link_state, link->link_status); etcp_inflight_nullify_link(link->etcp->input_wait_ack, link); etcp_inflight_nullify_link(link->etcp->input_send_q, link); - etcp_on_link_down(link->etcp, link); + etcp_fire_link_status_cbk(link, old_state, old_status); + if (!etcp->close_requested) etcp_on_link_down(etcp, NULL); callring_put(CR_LINK_CLOSE_FREE, (uintptr_t)link, 0); u_free(link->bbr); u_free(link); + etcp->callbacks_running--; } // ====== TCP link helpers ====== @@ -1291,17 +1284,26 @@ static void tcp_send_observed_ip(struct ETCP_LINK *link) { memory_pool_free(inst->pkt_pool, pkt); } +/* STCP-handle живёт до следующего poll; его отвязка обнаруживает удаление/замену линка, включая переиспользование адреса. */ +static int tcp_link_attached(struct ETCP_CONN *conn, struct ETCP_LINK *link, struct stcp_link *transport) { + if (conn->close_requested || conn->state == 2) return 0; + if (transport && stcp_link_get_etcp_link(transport) != link) return 0; + return etcp_link_rx_alive(conn, link) && !link->closing && link->tcp_link == transport; +} + // Поднять TCP-линк в ready: синхронизация reset-эпохи (gop/rid), keepalive, ed25519, // mark initialized + link_status=1, уведомить коннект и loadbalancer. void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) { - if (!link || !link->etcp) return; + if (!link || !link->etcp || link->closing || link->etcp->close_requested) return; struct ETCP_CONN *etcp = link->etcp; + struct stcp_link *transport = link->tcp_link; + etcp->callbacks_running++; /* reinit только при смене reset_id пира — синхронизация через etcp_conn_apply_peer_reset_id */ if (link->tcp_link) { uint64_t peer_rid = stcp_link_get_peer_reset_id(link->tcp_link); etcp_conn_apply_peer_reset_id(etcp, peer_rid); - if (etcp->state == 2) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "enter_ready_tcp: conn DELETED after apply_reset, link=%p", (void*)link); return; } + if (!tcp_link_attached(etcp, link, transport)) goto done; link->peer_device_type = stcp_link_get_peer_device_type(link->tcp_link); { uint16_t peer_ka = stcp_link_get_peer_keepalive_interval(link->tcp_link); @@ -1309,25 +1311,32 @@ void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) { etcp->instance->keepalive_interval, link->peer_device_type, peer_ka); } } - int old_state = link->link_state; + int old_state = link->link_state, old_status = link->link_status; link->initialized = 1; link->link_state = LINK_STATE_CONNECTED; link->link_status = 1; link->recv_keepalive = 1; link->last_recv_local_time = get_time_tb(); if (!link->mtu_remote) link->mtu_remote = link->mtu; - etcp->tcp_link_count++; + link->tcp_reconnect_delay_ms = 0; if (link->tcp_link) { const uint8_t* ed = stcp_link_get_peer_ed25519_pubkey(link->tcp_link); if (ed) memcpy(etcp->peer_ed25519_pubkey, ed, SC_PUBKEY_SIZE); } // Известная эпоха пира не подтверждает новый поток; следующий линк не перезапускает согласование. if (!etcp->initialized && !etcp->session_required) etcp_session_start(etcp); + if (!tcp_link_attached(etcp, link, transport)) goto done; if (etcp->initialized == 0) etcp_conn_ready(etcp); - if (etcp->state == 2) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "enter_ready_tcp: conn DELETED after conn_ready, link=%p", (void*)link); return; } + if (!tcp_link_attached(etcp, link, transport)) goto done; etcp_keepalive_link_ready(link); + if (!tcp_link_attached(etcp, link, transport)) goto done; loadbalancer_link_ready(link); - etcp_fire_link_status_cbk(link, old_state, link->link_status); - DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d UP (mtu=%d init=%d up=%d tcp_links=%d)", + if (!tcp_link_attached(etcp, link, transport)) goto done; + etcp_fire_link_status_cbk(link, old_state, old_status); + if (!tcp_link_attached(etcp, link, transport)) goto done; + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d UP (mtu=%d init=%d up=%d)", etcp->log_name, link->local_link_id, link->mtu, - etcp->initialized, etcp->links_up, etcp->tcp_link_count); + etcp->initialized, etcp->links_up); if (etcp->instance->nat_det) nat_detection_link_ready(etcp->instance->nat_det, link); + if (!tcp_link_attached(etcp, link, transport)) goto done; tcp_send_observed_ip(link); +done: + etcp->callbacks_running--; } // Повторная попытка STCP-подключения с экспоненциальным backoff (до 30 с). @@ -1335,24 +1344,27 @@ static void tcp_link_reconnect_cb(void *arg) { struct ETCP_LINK *link = (struct ETCP_LINK *)arg; if (!link || !link->etcp || !link->etcp->instance) return; link->tcp_reconnect_timer = NULL; - if (link->tcp_reconnect_delay_ms == 0) link->tcp_reconnect_delay_ms = 1000; + if (link->closing || link->etcp->close_requested || link->etcp->state == 2) return; DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d reconnect attempt (delay=%ums)", link->etcp->log_name, link->local_link_id, link->tcp_reconnect_delay_ms); - uint16_t port = ntohs(((struct sockaddr_in *)&link->remote_addr)->sin_port); - struct stcp_link *sl = stcp_link_connect(link, &link->remote_addr, port); - if (!sl) { link->tcp_reconnect_delay_ms *= 2; if (link->tcp_reconnect_delay_ms > 30000) link->tcp_reconnect_delay_ms = 30000; - link->tcp_reconnect_timer = uasync_set_timeout(link->etcp->instance->ua, (int)(link->tcp_reconnect_delay_ms * 10), link, tcp_link_reconnect_cb, "tcp_rct"); return; } - link->tcp_link = sl; - stcp_link_set_on_close(sl, tcp_link_close_cb, link); + etcp_tcp_link_start_connect(link, &link->remote_addr, ss_get_port(&link->remote_addr)); } // Исходящее TCP-подключение линка: stcp_link_connect + установка close-callback. void etcp_tcp_link_start_connect(struct ETCP_LINK *link, struct sockaddr_storage *addr, uint16_t port) { - if (!link || !link->etcp || !addr) return; + if (!link || !link->etcp || !addr || link->closing || link->etcp->close_requested || link->etcp->state == 2) return; + if (link->tcp_link || link->tcp_reconnect_timer) { + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] TCP connect already pending path=%u", link->etcp->log_name, link->local_link_id); + return; + } /* link->conn опционален: NULL = без локального bind (ОС выбирает source IP/интерфейс). * Задан — bind к interface_addr этого сокета (выбор интерфейса/ip). */ - memcpy(&link->remote_addr, addr, sizeof(*addr)); + if (addr != &link->remote_addr) memcpy(&link->remote_addr, addr, sizeof(*addr)); + link->link_state = LINK_STATE_HANDSHAKE; struct stcp_link *sl = stcp_link_connect(link, &link->remote_addr, port); - if (!sl) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_tcp_link_start_connect: stcp_link_connect failed"); return; } + if (!sl) { + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] TCP connect startup failed path=%u", link->etcp->log_name, link->local_link_id); + etcp_tcp_link_start_reconnect(link); return; + } link->tcp_link = sl; if (link->is_server == 0) stcp_link_set_on_close(sl, tcp_link_close_cb, link); DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d → stcp_link_connect %s:%u rc=%p bind=%s", @@ -1363,25 +1375,39 @@ void etcp_tcp_link_start_connect(struct ETCP_LINK *link, struct sockaddr_storage // Запланировать реконнект TCP-линка после обрыва (через tcp_reconnect_timer). void etcp_tcp_link_start_reconnect(struct ETCP_LINK *link) { if (!link || !link->etcp || !link->etcp->instance || link->is_server) return; - if (link->etcp->state == 2) return; + if (link->closing || link->etcp->close_requested || link->etcp->state == 2) return; if (link->tcp_reconnect_timer) return; if (link->tcp_link) { stcp_link_close(link->tcp_link); link->tcp_link = NULL; } + link->initialized = link->recv_keepalive = link->link_status = 0; link->link_state = LINK_STATE_HANDSHAKE; - if (link->etcp->tcp_link_count > 0) link->etcp->tcp_link_count--; + uint32_t previous = link->tcp_reconnect_delay_ms; + link->tcp_reconnect_delay_ms = !previous ? 1000 : previous >= 15000 ? 30000 : previous * 2; link->tcp_reconnect_timer = uasync_set_timeout(link->etcp->instance->ua, (int)(link->tcp_reconnect_delay_ms * 10), link, tcp_link_reconnect_cb, "tcp_rct"); + if (!link->tcp_reconnect_timer) { + struct ETCP_CONN *conn = link->etcp; + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] TCP reconnect timer allocation failed path=%u", conn->log_name, link->local_link_id); + conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--; + return; + } + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TCP reconnect scheduled path=%u delay_ms=%u previous_ms=%u", + link->etcp->log_name, link->local_link_id, link->tcp_reconnect_delay_ms, previous); } // Callback закрытия STCP-линка: серверный линк закрываем, клиентский — отправляем на реконнект. static void tcp_link_close_cb(struct stcp_link *sl, int err, void *arg) { struct ETCP_LINK *link = (struct ETCP_LINK *)arg; callring_put(CR_TCPCB_IN, (uintptr_t)sl, (uintptr_t)arg); - if (!link || !link->etcp) return; - if (link->etcp->state == 2) return; + if (!link || !link->etcp || link->closing) return; + if (link->tcp_link != sl) { + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] stale TCP close ignored path=%u", link->etcp->log_name, link->local_link_id); + stcp_link_close(sl); return; + } /* Сохраняем до каскада: etcp_on_link_down может закрыть conn и освободить link */ - int old_state = link->link_state; + int old_state = link->link_state, old_status = link->link_status; int is_server = link->is_server; int local_link_id = link->local_link_id; struct ETCP_CONN *etcp = link->etcp; + etcp->callbacks_running++; stcp_link_close(sl); link->tcp_link = NULL; @@ -1395,11 +1421,15 @@ static void tcp_link_close_cb(struct stcp_link *sl, int err, void *arg) { callring_put(CR_TCPCB_FREE, (uintptr_t)link, 0); etcp_link_close(link); } else { - etcp_fire_link_status_cbk(link, old_state, 0); + link->initialized = link->recv_keepalive = link->link_status = 0; DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d down err=%d, scheduling reconnect", etcp->log_name, local_link_id, err); - etcp_on_link_down(etcp, link); - if (etcp->state != 2) etcp_tcp_link_start_reconnect(link); + etcp_tcp_link_start_reconnect(link); + if (!etcp->close_requested && (old_state != link->link_state || old_status)) + etcp_fire_link_status_cbk(link, old_state, old_status); + /* Callback мог удалить link или соседний путь; агрегат пересчитывается только по живому списку. */ + if (!etcp->close_requested) etcp_on_link_down(etcp, NULL); } + etcp->callbacks_running--; } // Единый plaintext для UDP/STCP. Эпохи DATA/ACK подписываются вместе с содержимым.