From b73d6e91e2f89d7e813ba1a753631c1978bfc08c Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 23 Oct 2026 19:23:30 +0200 Subject: [PATCH] Verify compact attempt isolation, wire boundaries and exhaustion recovery --- src/transport_layer/etcp_tx.c | 32 +++++++------- tests/test_etcp_windows.c | 82 ++++++++++++++++++++++++++++++++++- 2 files changed, 97 insertions(+), 17 deletions(-) diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index bd6f9e2d..494e7f64 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -53,7 +53,7 @@ static void attempt_retire(struct ETCP_LINK* link, struct ETCP_TX_ATTEMPT* attem struct INFLIGHT_PACKET* packet = logical_find(conn, attempt->seq); if (!conn->active_attempts || link->inflight_bytes < attempt->ll.len || !link->inflight_packets || (packet && !packet->active_attempts)) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] attempt accounting mismatch link=%u id=%llu seq=%u flight=%u/%u len=%u logical=%u", + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] attempt accounting mismatch link=%u key=%llu seq=%u flight=%u/%u len=%u logical=%u", conn->log_name, link->local_link_id, (unsigned long long)attempt->attempt_id, attempt->seq, link->inflight_bytes, link->inflight_packets, attempt->ll.len, packet ? packet->active_attempts : 0); abort(); @@ -65,8 +65,8 @@ static void attempt_retire(struct ETCP_LINK* link, struct ETCP_TX_ATTEMPT* attem link->bbr_loss_since_ack = lost > INT_MAX ? INT_MAX : (uint32_t)lost; bbr_note_loss(link->bbr); } - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] retire link=%u attempt=%llu seq=%u reason=%s flight=%u/%u logical_active=%u", - conn->log_name, link->local_link_id, (unsigned long long)attempt->attempt_id, attempt->seq, + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] retire link=%u attempt=%u seq=%u reason=%s flight=%u/%u logical_active=%u", + conn->log_name, link->local_link_id, (unsigned)(uint16_t)attempt->attempt_id, attempt->seq, loss ? "RTO" : retry ? "link_close" : "ACK/rollback", link->inflight_bytes, link->inflight_packets, packet ? packet->active_attempts : 0); queue_remove_data(link->attempts, &attempt->ll); queue_entry_free(&attempt->ll); @@ -91,8 +91,8 @@ static void retransmit_cb(void* arg) { while (link->attempts->head) { struct ETCP_TX_ATTEMPT* attempt = (struct ETCP_TX_ATTEMPT*)link->attempts->head; if (!attempt->sent_at_us || now - attempt->sent_at_us < rto) break; - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RTO link=%u attempt=%llu seq=%u age_us=%llu RTO_us=%u RTT_us=%u jitter_us=%u", - link->etcp->log_name, link->local_link_id, (unsigned long long)attempt->attempt_id, attempt->seq, + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RTO link=%u attempt=%u seq=%u age_us=%llu RTO_us=%u RTT_us=%u jitter_us=%u", + link->etcp->log_name, link->local_link_id, (unsigned)(uint16_t)attempt->attempt_id, attempt->seq, (unsigned long long)(now - attempt->sent_at_us), rto, link->data_rtt_us, link->data_jitter_us); attempt_retire(link, attempt, 1, 1); } @@ -203,8 +203,8 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) { attempt->is_app_limited = !conn->input_queue->count && conn->input_send_q->count == 1 && (!pn || (!pn->input->count && !pn->pending && !pn->data_ptr)); if (link->bbr && link->bbr->mode == BBR_PROBE_RTT) attempt->is_app_limited = 1; - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] sent seq=%u attempt=%llu path=%u previous=%u bytes=%u flight=%u/%u", - conn->log_name, packet->seq, (unsigned long long)attempt->attempt_id, link->local_link_id, + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] sent seq=%u attempt=%u path=%u previous=%u bytes=%u flight=%u/%u", + conn->log_name, packet->seq, (unsigned)(uint16_t)attempt->attempt_id, link->local_link_id, packet->last_link ? packet->last_link->local_link_id : 0, packet->ll.len, link->inflight_bytes, link->inflight_packets); if (packet->last_attempt_id && packet->last_link) packet->last_link->total_retransmissions++; packet->send_hist[packet->send_count % 8] = link->local_link_id; @@ -227,7 +227,7 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint } struct ETCP_TX_ATTEMPT* attempt = (struct ETCP_TX_ATTEMPT*)queue_find_data_by_index(link->attempts, &attempt_id); if (attempt && (attempt->seq != seq || !attempt->sent_at_us)) { - DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] invalid attempt ACK path=%u id=%llu seq=%u expected=%u committed=%d", + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] invalid attempt ACK path=%u key=%llu seq=%u expected=%u committed=%d", conn->log_name, link->local_link_id, (unsigned long long)attempt_id, seq, attempt->seq, !!attempt->sent_at_us); return; } @@ -238,8 +238,8 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint ((peer_rx_bytes > link->peer_rx_data_bytes && peer_rx_time_us < link->peer_rx_data_time_us) || (peer_rx_bytes < link->peer_rx_data_bytes && peer_rx_time_us > link->peer_rx_data_time_us) || (peer_rx_bytes == link->peer_rx_data_bytes && peer_rx_time_us != link->peer_rx_data_time_us)))) { - DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] inconsistent RX point path=%u attempt=%llu point=%llu/%llu previous=%llu/%llu", - conn->log_name, link->local_link_id, (unsigned long long)attempt_id, + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] inconsistent RX point path=%u seq=%u attempt=%u point=%llu/%llu previous=%llu/%llu", + conn->log_name, link->local_link_id, seq, (unsigned)(uint16_t)attempt_id, (unsigned long long)peer_rx_bytes, (unsigned long long)peer_rx_time_us, (unsigned long long)link->peer_rx_data_bytes, (unsigned long long)link->peer_rx_data_time_us); rx_valid = 0; @@ -286,8 +286,8 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint } } if (interval > UINT32_MAX || (interval && !rs.interval_us)) - DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] BW interval unrepresentable path=%u attempt=%llu local_us=%llu RX=%llu/%llu", - conn->log_name, link->local_link_id, (unsigned long long)attempt_id, (unsigned long long)interval, + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] BW interval unrepresentable path=%u seq=%u attempt=%u local_us=%llu RX=%llu/%llu", + conn->log_name, link->local_link_id, seq, (unsigned)(uint16_t)attempt_id, (unsigned long long)interval, (unsigned long long)rx_bytes, (unsigned long long)rx_us); link->last_ack_time_us = now; if (sample.sent_at_us > link->first_tx_time_us) link->first_tx_time_us = sample.sent_at_us; @@ -296,17 +296,17 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint bbr_main(link->bbr, &rs, &cwnd, &pacing, conn->normalizer->frag_size, link->inflight_bytes, limited); etcp_link_update_inflight_lim(link, cwnd); etcp_loadbalancer_set_pacing_rate(link, pacing); } - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] ACK seq=%u attempt=%llu path=%u RTT_us=%u delay_us=%u sample=%u/%u " + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] ACK seq=%u attempt=%u path=%u RTT_us=%u delay_us=%u sample=%u/%u " "local_interval_us=%llu RX=%llu/%llu flight=%u cwnd=%u pace=%u RTO_us=%u", - conn->log_name, seq, (unsigned long long)attempt_id, link->local_link_id, + conn->log_name, seq, (unsigned)(uint16_t)attempt_id, link->local_link_id, rtt, receiver_delay_us, rs.delivered, rs.interval_us, (unsigned long long)interval, (unsigned long long)rx_bytes, (unsigned long long)rx_us, link->inflight_bytes, link->inflight_lim_bytes, link->bbr_pacing_rate, etcp_tx_rto_us(link)); link->acked_bytes += sample.ll.len; link->acked_packets++; link->bbr_loss_since_ack = 0; link->cwnd_limited = 0; retransmit_arm(link); - } else DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] old/foreign ACK seq=%u attempt=%llu path=%u: logical only", - conn->log_name, seq, (unsigned long long)attempt_id, link->local_link_id); + } else DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] old/foreign ACK seq=%u attempt=%u path=%u: logical only", + conn->log_name, seq, (unsigned)(uint16_t)attempt_id, link->local_link_id); etcp_ack_recv(conn, seq); etcp_conn_wake_sender(conn); } diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index 1b5aae03..7a800832 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -1001,6 +1001,85 @@ static void restart_allocation_failures(void) { } } +/* Проходит настоящий wire decoder: номер может совпадать у разных seq, нулевой запрещён. */ +static void receive_attempt_number(struct ETCP_LINK* link, uint32_t seq, uint16_t number) { + struct ETCP_DGRAM* pkt = memory_pool_alloc(inst.pkt_pool); CHECK(pkt); + pkt->link = link; pkt->data_len = etcp_ack_section_size(1); + memset(pkt->data,0,pkt->data_len); pkt->data[0] = ETCP_SECTION_ACK; pkt->data[1] = 1; + for (unsigned i = 0; i < 4; i++) pkt->data[ETCP_ACK_BASE_SIZE+i] = seq >> (8*i); + pkt->data[ETCP_ACK_BASE_SIZE+4] = number; pkt->data[ETCP_ACK_BASE_SIZE+5] = number >> 8; + etcp_conn_input(pkt); check_accounting(); +} + +static void compact_attempt_identity(void) { + setup(); enable_models(); fill(a,2,100); b->link_status = 0; + struct INFLIGHT_PACKET* first = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head; + struct INFLIGHT_PACKET* second = (struct INFLIGHT_PACKET*)first->ll.next; + CHECK(first && second && first->seq == 1 && second->seq == 2); + CHECK(first->last_attempt_id == etcp_tx_attempt_id(1,1) && second->last_attempt_id == etcp_tx_attempt_id(2,1)); + CHECK(last_attempt(first)->seq == 1 && last_attempt(second)->seq == 2 && a->attempts->count == 2); + conn->got_initial_pkt = 1; + receive_fragment(a,2,1); receive_fragment(a,3,1); receive_fragment(a,2,1); + CHECK(a->ack_q->count == 2 && a->rx_data_bytes == 3); + struct ETCP_DGRAM* pkt = request(); CHECK(pkt && !pkt->payload_len && pkt->data[1] == 2); + const uint8_t entry1[] = {2,0,0,0,1,0}, entry2[] = {3,0,0,0,1,0}; + CHECK(!memcmp(pkt->data+ETCP_ACK_BASE_SIZE,entry1,sizeof(entry1))); + CHECK(!memcmp(pkt->data+ETCP_ACK_BASE_SIZE+ETCP_ACK_ENTRY_SIZE,entry2,sizeof(entry2))); + CHECK(pkt->data_len == etcp_ack_section_size(2)); memory_pool_free(inst.pkt_pool,pkt); + receive_attempt_number(a,1,0); CHECK(a->inflight_bytes == 200 && conn->unacked_bytes == 200 && !a->bbr->delivered); + receive_attempt_number(b,1,1); CHECK(a->inflight_bytes == 200 && conn->input_wait_ack->count == 1); + CHECK(conn->unacked_bytes == 100 && !a->bbr->delivered && !b->bbr->delivered); + receive_attempt_number(a,1,1); CHECK(a->inflight_bytes == 100 && a->bbr->delivered == 100 && !b->bbr->delivered); + receive_attempt_number(a,1,1); CHECK(a->inflight_bytes == 100 && a->bbr->delivered == 100); + receive_attempt_number(a,2,1); CHECK(!a->inflight_bytes && !conn->unacked_bytes && a->bbr->delivered == 200); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG,"Compact attempts: same number on different seq stays isolated; zero/foreign/duplicate ACK checked"); + teardown(); +} + +/* Номер расходуется даже при отказе send; 65535 передаётся полностью, следующая попытка меняет эпоху. */ +static void compact_attempt_exhaustion(void) { + setup(); b->link_status = 0; + struct INFLIGHT_PACKET* packet = enqueue(1); + struct ETCP_DGRAM* pkt = etcp_request_pkt(conn); CHECK(pkt && pkt->attempt); + CHECK(pkt->attempt->attempt_id == etcp_tx_attempt_id(1,1)); + etcp_tx_complete(pkt,0); memory_pool_free(inst.pkt_pool,pkt); + CHECK(packet->attempt_count == 1 && !packet->last_attempt_id && !packet->send_count && !a->inflight_packets); + pkt = request(); CHECK(pkt && packet->attempt_count == 2 && packet->last_attempt_id == etcp_tx_attempt_id(1,2)); + memory_pool_free(inst.pkt_pool,pkt); ack_all(); + packet = enqueue(1); packet->attempt_count = UINT16_MAX-1; + pkt = request(); CHECK(pkt && packet->attempt_count == UINT16_MAX); + const uint8_t last_data[] = {ETCP_SECTION_PAYLOAD,2,0,0,0,255,255,0x5a}; + CHECK(pkt->data_len == ETCP_ACK_BASE_SIZE+sizeof(last_data)); + CHECK(!memcmp(pkt->data+ETCP_ACK_BASE_SIZE,last_data,sizeof(last_data))); + CHECK(packet->last_attempt_id == etcp_tx_attempt_id(2,UINT16_MAX)); memory_pool_free(inst.pkt_pool,pkt); + conn->initialized = 0; test_clock_run_for(ua,120); conn->initialized = 1; + CHECK(!a->inflight_packets && packet->state == INFLIGHT_STATE_WAIT_SEND); + uint64_t epoch = conn->reset_id; + CHECK(!request() && conn->reset_id != epoch && conn->reinit_pending && conn->session_required); + CHECK(!conn->active_attempts && !conn->unacked_bytes && !conn->input_send_q->count && !conn->input_wait_ack->count); + confirm_test_session(); packet = enqueue(1); pkt = request(); CHECK(pkt); + CHECK(packet->seq == 1 && packet->attempt_count == 1 && packet->last_attempt_id == etcp_tx_attempt_id(1,1)); + memory_pool_free(inst.pkt_pool,pkt); ack_all(); teardown(); +} + +/* Полный круг seq также не должен повторять пару в прежней эпохе, даже с первым номером попытки. */ +static void compact_sequence_exhaustion(void) { + setup(); b->link_status = 0; + conn->next_tx_id = UINT32_MAX; conn->rx_ack_till = UINT32_MAX-1; + const uint8_t normalized[] = {1,0,253}; + CHECK(etcp_int_send(conn,normalized,sizeof(normalized)) == 0 && !conn->next_tx_id); + struct ETCP_DGRAM* pkt = request(); CHECK(pkt); + CHECK(pkt->attempt == NULL && a->attempts->count == 1); + struct ETCP_TX_ATTEMPT* attempt = (struct ETCP_TX_ATTEMPT*)a->attempts->head; + CHECK(attempt->seq == UINT32_MAX && attempt->attempt_id == etcp_tx_attempt_id(UINT32_MAX,1)); + memory_pool_free(inst.pkt_pool,pkt); + uint64_t epoch = conn->reset_id; + CHECK(etcp_int_send(conn,normalized,sizeof(normalized)) == 0); + CHECK(conn->reset_id != epoch && conn->reinit_pending && conn->session_required && conn->next_tx_id == 1); + CHECK(!conn->active_attempts && !conn->unacked_bytes && !conn->input_queue->count); + confirm_test_session(); teardown(); +} + int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_DEBUG); @@ -1012,7 +1091,8 @@ int main(void) { retransmission_timeout, pacing_rate_changes, pacing_timer_failure, retry_rotation, independent_attempt_acks, receive_admission_and_path_acks, peer_window_queries, pacing_continuity, pacing_data_units, attempt_memory_limits, normalizer_rx_oom, normalizer_tx_oom, normalizer_callback_abort, reset_without_allocation, restart_allocation_failures, - receiver_bandwidth_bounds, receiver_point_lifecycle, receiver_point_wire, receiver_point_truncated}; + receiver_bandwidth_bounds, receiver_point_lifecycle, receiver_point_wire, receiver_point_truncated, + compact_attempt_identity, compact_attempt_exhaustion, compact_sequence_exhaustion}; for (unsigned i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { cases[i](); CHECK(u_get_allocated_count() == baseline); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS ETCP window scenario %u", i);