Retain ETCP links across callbacks and release closed paths on scope exit

src/transport_layer/etcp_connections.c

@@ -65,7 +65,6 @@ 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); }
@@ -353,27 +352,32 @@ void etcp_link_restart_init_timer(struct ETCP_LINK* link) {
 // Клиент: переход в handshake — отправить INIT и запустить таймер повторов.
 void etcp_link_enter_init(struct ETCP_LINK* link) {
     DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
-    if (!link) return;
+    if (!link || link->closing || link->etcp->close_requested) return;
     struct ETCP_CONN *conn = link->etcp;
+    conn->callbacks_running++; link->active_users++;
     int old_state = link->link_state;
     link->link_state = LINK_STATE_HANDSHAKE; // handshake
     etcp_fire_link_status_cbk(link, old_state, link->link_status);
-    if (!etcp_link_rx_alive(conn, link)) return;
-    if (link->is_server != 0) return;
+    if (!etcp_link_is_alive(conn, link) || link->is_server != 0) goto done;
     int reset = (link->etcp->got_initial_pkt == 0);
     etcp_link_send_init(link, reset, 0);
     etcp_link_restart_init_timer(link);
+done:
+    etcp_link_release(link); conn->callbacks_running--;
 }
 
 // Клиент: повторный handshake после разрыва — уведомить о падении, отправить INIT, снять keepalive.
 void etcp_link_enter_reinit(struct ETCP_LINK* link) {
     DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
-    if (!link) return;
+    if (!link || link->closing || link->etcp->close_requested) return;
+    struct ETCP_CONN *conn = link->etcp;
+    conn->callbacks_running++; link->active_users++;
     int old_state = link->link_state;
     link->link_state = LINK_STATE_TRY_RECONNECT; // reconnect
     etcp_fire_link_status_cbk(link, old_state, link->link_status);
-    etcp_on_link_down(link->etcp, link);
-    if (link->is_server != 0) return;
+    if (!etcp_link_is_alive(conn, link)) goto done;
+    etcp_on_link_down(conn, link);
+    if (!etcp_link_is_alive(conn, link) || link->is_server != 0) goto done;
     int reset = (link->etcp->got_initial_pkt == 0);
     etcp_link_send_init(link, reset, 0);
 
@@ -383,6 +387,8 @@ void etcp_link_enter_reinit(struct ETCP_LINK* link) {
     }
 
     etcp_link_restart_init_timer(link);
+done:
+    etcp_link_release(link); conn->callbacks_running--;
 }
 
 
@@ -1163,25 +1169,26 @@ struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn
     DEBUG_INFO(DEBUG_CATEGORY_ETCP, "NEW link etcp=[%s] link=%p socket=%s id=%u server=%u mtu=%u tcp=%u DATA_MSS=%u cwnd=%u",
         etcp->log_name, link, conn ? conn->name : "tcp", link->local_link_id, link->is_server, link->mtu, link->is_tcp,
         etcp->normalizer->frag_size, link->inflight_lim_bytes);
-    link->constructing = 1;
+    link->active_users++;
     etcp_fire_link_status_cbk(link, -1, 0);
 
     if (!link->closing && !etcp->close_requested && !is_tcp && is_server == 0) {
         etcp_link_enter_init(link);
     }
-    link->constructing = 0;
     if (link->closing) {
         DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] link creation cancelled path=%u", etcp->log_name, link->local_link_id);
-        u_free(link->bbr); u_free(link);
+        etcp_link_release(link);
         return NULL;
     }
     if (etcp->close_requested) {
         DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] link creation stopped by connection close path=%u", etcp->log_name, link->local_link_id);
         etcp_link_close(link);
+        etcp_link_release(link);
         return NULL;
     }
 
     callring_put(CR_LINK_NEW, (uintptr_t)link, (uintptr_t)etcp);
+    etcp_link_release(link);
     return link;
 }
 
@@ -1210,11 +1217,24 @@ static void etcp_inflight_nullify_link(struct ll_queue* q, struct ETCP_LINK* dea
     }
 }
 
+void etcp_link_release(struct ETCP_LINK* link) {
+    if (!link || !link->active_users) {
+        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "link release without active owner link=%p", (void*)link);
+        abort();
+    }
+    link->active_users--;
+    if (link->closing && !link->active_users) {
+        DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] closed link memory released path=%u", link->etcp->log_name, link->local_link_id);
+        u_free(link->bbr); u_free(link);
+    }
+}
+
 // Закрытие линка: отмена таймеров (init/shaper/keepalive/tcp), удаление из списков и очереди,
 // зануление ссылок inflight (защита от use-after-free) и освобождение памяти.
 void etcp_link_close(struct ETCP_LINK* link) {
     DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
     if (!link || link->closing) return;
+    link->active_users++;
     link->closing = 1;
     struct ETCP_CONN *etcp = link->etcp;
     etcp->callbacks_running++;
@@ -1279,7 +1299,7 @@ void etcp_link_close(struct ETCP_LINK* 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);
-    if (!link->constructing) { u_free(link->bbr); u_free(link); }
+    etcp_link_release(link);
     etcp->callbacks_running--;
 }
 
@@ -1312,7 +1332,7 @@ static void tcp_send_observed_ip(struct ETCP_LINK *link) {
 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;
+    return etcp_link_is_alive(conn, link) && link->tcp_link == transport;
 }
 
 // Поднять TCP-линк в ready: синхронизация reset-эпохи (gop/rid), keepalive, ed25519,
@@ -1322,6 +1342,7 @@ void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) {
     struct ETCP_CONN *etcp = link->etcp;
     struct stcp_link *transport = link->tcp_link;
     etcp->callbacks_running++;
+    link->active_users++;
 
     /* reinit только при смене reset_id пира — синхронизация через etcp_conn_apply_peer_reset_id */
     if (link->tcp_link) {
@@ -1360,6 +1381,7 @@ void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) {
     if (!tcp_link_attached(etcp, link, transport)) goto done;
     tcp_send_observed_ip(link);
 done:
+    etcp_link_release(link);
     etcp->callbacks_running--;
 }
 
@@ -2045,8 +2067,8 @@ static int etcp_link_record_rx(struct ETCP_LINK* link, const struct ETCP_DGRAM*
 }
 
 /* Проверка заимствованного link после callback: conn удерживается вызывающим, сам link мог быть удалён. */
-static int etcp_link_rx_alive(struct ETCP_CONN* conn, struct ETCP_LINK* link) {
-    if (conn->close_requested) return 0;
+int etcp_link_is_alive(struct ETCP_CONN* conn, struct ETCP_LINK* link) {
+    if (conn->close_requested || link->closing) return 0;
     for (struct ETCP_LINK* current = conn->links; current; current = current->next) if (current == link) return 1;
     return 0;
 }
@@ -2058,7 +2080,7 @@ static int etcp_link_publish_rx(struct ETCP_LINK* link, int old_state, int old_s
     if (conn->close_requested) return 0;
     if (old_state != link->link_state || old_status != link->link_status)
         etcp_fire_link_status_cbk(link, old_state, old_status);
-    if (!etcp_link_rx_alive(conn, link)) return 0;
+    if (!etcp_link_is_alive(conn, link)) return 0;
     if (old_status != link->link_status)
         DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] RX link=%u status=%d->%u recv=%u remote=%u state=%u init=%u",
             conn->log_name, link->local_link_id, old_status, link->link_status, link->recv_keepalive,
@@ -2066,13 +2088,14 @@ static int etcp_link_publish_rx(struct ETCP_LINK* link, int old_state, int old_s
     if (old_status && !link->link_status) etcp_on_link_down(conn, link);
     else if (link->initialized && link->link_status && (old_status != link->link_status || old_state != link->link_state))
         loadbalancer_link_ready(link);
-    return etcp_link_rx_alive(conn, link);
+    return etcp_link_is_alive(conn, link);
 }
 
 // Сервер: принять RX INIT, отправить ответ и запустить keepalive даже при неподтверждённом обратном пути.
 static void send_init_response(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, struct ETCP_CONN* conn, const struct ETCP_INIT_REQUEST_PKT* req, const struct sockaddr_storage* addr, uint8_t send_reset, uint32_t req_src_ip, uint16_t req_src_port, size_t pkt_len) {
     if (conn->close_requested) { memory_pool_free(e_sock->instance->pkt_pool, pkt); return; }
     conn->callbacks_running++;
+    link->active_users++;
     int old_state = link->link_state, old_status = etcp_link_record_rx(link, pkt);
     uint8_t client_socket_id = req->socket_id;
     link->is_server = 1;
@@ -2144,14 +2167,14 @@ static void send_init_response(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pk
 
     memory_pool_free(e_sock->instance->pkt_pool, pkt);
     if (!conn->initialized) etcp_conn_ready(conn);
-    if (etcp_link_rx_alive(conn, link)) {
+    if (etcp_link_is_alive(conn, link)) {
         DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] INIT accepted link=%u status=%u recv=%u remote=%u", conn->log_name,
             link->local_link_id, link->link_status, link->recv_keepalive, link->remote_keepalive);
         etcp_keepalive_link_ready(link);
         if (etcp_link_publish_rx(link, old_state, old_status) && conn->instance->nat_det)
             nat_detection_link_ready(conn->instance->nat_det, link);
     }
-    conn->callbacks_running--;
+    etcp_link_release(link); conn->callbacks_running--;
 }
 
 // Клиент: обработка INIT_RESPONSE — MTU, link_id, NAT-адрес, ed25519, keepalive, подъём линка.
@@ -2221,7 +2244,7 @@ static int handle_init_response_client(struct ETCP_SOCKET* e_sock, struct ETCP_D
 
     memory_pool_free(e_sock->instance->pkt_pool, pkt);
     if (!conn->initialized) etcp_conn_ready(conn);
-    if (etcp_link_rx_alive(conn, link)) {
+    if (etcp_link_is_alive(conn, link)) {
         DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] INIT_RESPONSE accepted link=%u status=%u recv=%u remote=%u mtu=%u",
             conn->log_name, link->local_link_id, link->link_status, link->recv_keepalive, link->remote_keepalive, link->mtu);
         etcp_keepalive_link_ready(link);
@@ -2692,6 +2715,7 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
     uint32_t reset = conn->reset_count;
     int result = 0;
     conn->callbacks_running++;
+    link->active_users++;
     if (pkt_len<3) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "decrypted packet too small, size=%zu", pkt_len); result = 46; goto done; }
     pkt->data_len=pkt_len-3;
     pkt->noencrypt_len=0;
@@ -2785,7 +2809,7 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
         memory_pool_free(link->etcp->instance->pkt_pool, pkt);
     }
 done:
-    conn->callbacks_running--;
+    etcp_link_release(link); conn->callbacks_running--;
     return result;
 }
 

src/transport_layer/etcp_connections.h

@@ -232,7 +232,7 @@ struct ETCP_LINK {
     uint8_t is_server;              // инициирует подключение клиент
     uint8_t initialized;            // Флаг инициализации (1=подтверждено или получен request)
     uint8_t closing;                // Close начат: повторный вход из callback игнорируется до окончательного free
-    uint8_t constructing;           // Конструктор ещё владеет памятью: close отвязывает ресурсы, free выполняет конструктор
+    uint32_t active_users;          // Внутренние стеки конструктора/RX/событий; память закрытого линка живёт до выхода последнего
     uint8_t local_link_id;          // id моего линка
     uint8_t remote_link_id;         // id этого линка на peer (устанавливается в момент initialized)
     uint8_t remote_socket_id;       // socket id peer
@@ -411,7 +411,7 @@ void tcp_server_on_link(struct stcp_link *link, struct ETCP_SOCKET *tcp_sock);
 
 /* Создать линк etcp через conn к remote_addr; conn=NULL выбирает исходящий TCP без bind, is_server задаёт роль.
  * Уведомляет о создании, для исходящего UDP запускает INIT. Ошибка/close из callback — NULL без живого линка.
- * До возврата память принадлежит конструктору: отмена закрывает ресурсы, освобождение завершает конструктор. */
+ * Конструктор удерживает память до возврата; отмена сразу закрывает ресурсы. */
 struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn,
                                 const struct sockaddr_storage* remote_addr, uint8_t is_server);
 
@@ -419,12 +419,19 @@ struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn
 // Минимум выбирает BBR; ноль отклоняется с логом без изменения окна. Неверный link игнорируется.
 void etcp_link_update_inflight_lim(struct ETCP_LINK* link, uint32_t new_lim);
 
-/* Снимает линк из списков, отменяет таймеры/STCP и освобождает память до возврата.
- * При вызове внутри его конструктора память сохраняется до выхода из конструктора, который вернёт NULL.
- * Из callback его закрытия повторный close допустим; после возврата handle недействителен.
+/* Снимает линк из списков и отменяет таймеры/STCP; handle сразу становится недействителен.
+ * Память освобождается после выхода всех активных внутренних обработчиков, иначе до возврата.
+ * Из callback его закрытия повторный close допустим; сохранять handle для последующих событий нельзя.
  * Callback может закрыть ETCP_CONN или соседний линк: вызывающий не должен хранить next через уведомление. */
 void etcp_link_close(struct ETCP_LINK* link);
 
+/* Внутренний release после active_users++: последний обработчик освобождает закрытый link.
+ * Возврат не гарантирует сохранность указателя. Новых таймеров/выделений памяти нет. */
+void etcp_link_release(struct ETCP_LINK* link);
+/* Проверить после callback, что conn не закрывается и тот же link ещё в его списке.
+ * Вызывающий удерживает conn через callbacks_running и link через active_users, исключая повторное использование адреса. */
+int etcp_link_is_alive(struct ETCP_CONN* conn, struct ETCP_LINK* link);
+
 //int etcp_input_cbk(struct packet_buffer* pkt, struct ETCP_SOCKET* conn);// получает расшифрованный пакет
 int etcp_encrypt_send(struct ETCP_DGRAM* dgram);// зашифровывает и отправляет пакет