Track stalled DATA paths and accepted retry rounds safely

src/transport_layer/etcp_connections.c

@@ -1205,14 +1205,14 @@ void etcp_link_update_inflight_lim(struct ETCP_LINK* link, uint32_t new_lim) {
     etcp_conn_on_inflight_lim_changed(link->etcp);
 }
 
-// Обнуляет last_link для всех inflight-записей ссылающихся на dead_link.
-// Вызывается перед u_free(link) чтобы etcp_ack_recv не упал на use-after-free.
+// Убрать курсор и бит слота до удаления линка из списка: callbacks смогут безопасно переиспользовать его local_link_id.
 static void etcp_inflight_nullify_link(struct ll_queue* q, struct ETCP_LINK* dead_link) {
     if (!q) return;
     struct ll_entry* entry = q->head;
     while (entry) {
         struct INFLIGHT_PACKET* inf = (struct INFLIGHT_PACKET*)entry;
         if (inf->last_link == dead_link) inf->last_link = NULL;
+        inf->used_links[dead_link->local_link_id >> 3] &= (uint8_t)~(1u << (dead_link->local_link_id & 7));
         entry = entry->next;
     }
 }
@@ -1278,6 +1278,9 @@ void etcp_link_close(struct ETCP_LINK* link) {
         link->keepalive_timer = NULL;
     }
 
+    etcp_inflight_nullify_link(etcp->input_wait_ack, link);
+    etcp_inflight_nullify_link(etcp->input_send_q, link);
+
     // универсальное удаление из односвязного списка
     struct ETCP_LINK **pp = &link->etcp->links;
     while (*pp) {
@@ -1294,8 +1297,6 @@ void etcp_link_close(struct ETCP_LINK* link) {
     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_inflight_nullify_link(link->etcp->input_wait_ack, link);
-    etcp_inflight_nullify_link(link->etcp->input_send_q, 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);

src/transport_layer/etcp_tx.c

@@ -5,6 +5,7 @@
 #include "../lib/debug_config.h"
 #include <limits.h>
 #include <stdlib.h>
+#include <string.h>
 
 static void retransmit_cb(void* arg);
 
@@ -60,6 +61,15 @@ static void attempt_retire(struct ETCP_LINK* link, struct ETCP_TX_ATTEMPT* attem
     }
     link->inflight_bytes -= attempt->ll.len; link->inflight_packets--; conn->active_attempts--;
     if (packet) packet->active_attempts--;
+    if (loss) {
+        if (link->data_rto_streak < ETCP_TX_STALL_THRESHOLD) link->data_rto_streak++;
+        if (!link->tx_stalled && link->data_rto_streak == ETCP_TX_STALL_THRESHOLD) {
+            link->tx_stalled = 1; link->tx_stalled_since_us = get_time_us();
+            DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] DATA path stalled link=%u consecutive_RTO=%u seq=%u flight=%u",
+                conn->log_name, link->local_link_id, link->data_rto_streak, attempt->seq, link->inflight_bytes);
+            etcp_conn_wake_sender(conn);
+        }
+    }
     if (loss && link->bbr) {
         uint64_t lost = (uint64_t)link->bbr_loss_since_ack + attempt->ll.len;
         link->bbr_loss_since_ack = lost > INT_MAX ? INT_MAX : (uint32_t)lost;
@@ -129,6 +139,9 @@ void etcp_tx_link_reset(struct ETCP_LINK* link, int retry) {
     link->shaper_load_time_tb = get_time_tb(); link->shaper_sub_nanotime = 0;
     link->bbr_pacing_rate = link->bandwidth = 0;
     link->data_rtt_us = link->data_jitter_us = link->bbr_loss_since_ack = 0;
+    if (link->tx_stalled) DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] clear stalled path on reset link=%u",
+        link->etcp->log_name, link->local_link_id);
+    link->data_rto_streak = link->tx_stalled = 0; link->tx_stalled_since_us = 0;
     link->ack_reply_pending = 0;
     link->delivered_bytes = link->acked_bytes = link->acked_packets = 0;
     link->rx_data_bytes = link->rx_data_time_us = 0;
@@ -209,6 +222,13 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) {
     if (packet->last_attempt_id && packet->last_link) packet->last_link->total_retransmissions++;
     packet->send_hist[packet->send_count % 8] = link->local_link_id;
     if (packet->send_count != UINT32_MAX) packet->send_count++;
+    if (etcp_packet_link_used(packet, link->local_link_id)) {
+        memset(packet->used_links, 0, sizeof(packet->used_links));
+        DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] retry starts new path round seq=%u link=%u", conn->log_name,
+            packet->seq, link->local_link_id);
+    }
+    packet->used_links[link->local_link_id >> 3] |= 1u << (link->local_link_id & 7);
+    conn->last_rr_link = link;
     packet->last_link = link; packet->last_timestamp = now / 100; packet->last_attempt_id = attempt->attempt_id;
     queue_remove_data(conn->input_send_q, &packet->ll); packet->state = INFLIGHT_STATE_WAIT_ACK;
     if (queue_data_put_with_index(conn->input_wait_ack, &packet->ll) < 0) {
@@ -233,6 +253,11 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint
     }
     if (attempt) {
         struct ETCP_TX_ATTEMPT sample = *attempt;
+        if (!link->tx_stalled || sample.sent_at_us >= link->tx_stalled_since_us) {
+            if (link->tx_stalled) DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] DATA path recovered link=%u seq=%u attempt=%u",
+                conn->log_name, link->local_link_id, seq, (unsigned)(uint16_t)attempt_id);
+            link->data_rto_streak = link->tx_stalled = 0; link->tx_stalled_since_us = 0;
+        }
         int rx_valid = peer_rx_bytes && peer_rx_time_us;
         if (!!peer_rx_bytes != !!peer_rx_time_us || (rx_valid && link->peer_rx_data_time_us &&
             ((peer_rx_bytes > link->peer_rx_data_bytes && peer_rx_time_us < link->peer_rx_data_time_us) ||