Exercise bounded attempt and ACK storage after logical delivery

src/transport_layer/etcp_tx.h

@@ -35,7 +35,9 @@ void etcp_tx_link_reset(struct ETCP_LINK* link, int retry);
 void etcp_tx_link_destroy(struct ETCP_LINK* link);
 /* Расчётный RTO в микросекундах: 2*(собственный RTT+jitter), 10мс..1с; до DATA ACK использует KA RTT/1с. */
 uint32_t etcp_tx_rto_us(const struct ETCP_LINK* link);
-/* Резервировать окно для packet; NULL — ошибка с логом, packet остаётся в очереди отправки. */
+/* Резервировать окно для packet; NULL — ошибка с логом, packet остаётся в очереди отправки.
+ * Физические записи ограничены MAX_INFLIGHT_SIZE на линк и 2*MAX_INFLIGHT_SIZE на соединение.
+ * Исчерпание бюджета блокирует выбор DATA до ACK/RTO/закрытия; ACK-очередь линка ограничена MAX_INFLIGHT_SIZE. */
 struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_PACKET* packet);
 /* Завершить синхронный send: success=1 принят транспортом, 0 — снять резерв без сетевой потери.
  * Вызвать ровно один раз до освобождения dgram и до следующего poll; dgram->attempt обнуляется. */

tests/test_etcp_windows.c

@@ -65,11 +65,14 @@ void __wrap_bbr_main(struct bbr* model, const struct bbr_rate_sample* sample, ui
 
 /* Счётчики сверяются с содержимым очередей, а не с ожидаемыми результатами самого алгоритма. */
 static void check_accounting(void) {
+    uint32_t attempts = 0;
     for (struct ETCP_LINK* l = conn->links; l; l = l->next) {
         uint32_t bytes = 0, packets = 0;
         for (struct ll_entry* e = l->attempts->head; e; e = e->next) { bytes += e->len; packets++; }
         CHECK(l->inflight_bytes == bytes && l->inflight_packets == packets);
+        attempts += packets;
     }
+    CHECK(conn->active_attempts == attempts);
     struct ll_queue* logical_queues[] = {conn->input_send_q, conn->input_wait_ack};
     for (unsigned i = 0; i < 2; i++) for (struct ll_entry* e = logical_queues[i]->head; e; e = e->next) {
         struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)e;
@@ -387,6 +390,36 @@ static void pacing_data_units(void) {
     teardown();
 }
 
+/* Физические копии уже доставленных seq также занимают бюджет, пока не придёт собственный ACK. */
+static void attempt_memory_limits(void) {
+    setup();
+    a->inflight_lim_bytes = b->inflight_lim_bytes = MAX_INFLIGHT_SIZE * 4;
+    for (unsigned path = 0; path < 2; path++) {
+        struct ETCP_LINK* link = path ? b : a;
+        for (unsigned i = 0; i < MAX_INFLIGHT_SIZE; i++) {
+            struct INFLIGHT_PACKET* packet = enqueue(1);
+            struct ETCP_TX_ATTEMPT* attempt = etcp_tx_prepare(link, packet); CHECK(attempt);
+            struct ETCP_DGRAM pkt = { .link = link, .attempt = attempt };
+            etcp_tx_complete(&pkt, 1); etcp_ack_recv(conn, packet->seq);
+        }
+        CHECK(link->attempts->count == MAX_INFLIGHT_SIZE);
+    }
+    CHECK(conn->active_attempts == 2 * MAX_INFLIGHT_SIZE);
+    struct INFLIGHT_PACKET* pending = enqueue(1);
+    CHECK(!etcp_loadbalancer_select_link(conn, 1, NULL));
+    CHECK(!etcp_tx_prepare(a, pending));
+    struct ETCP_TX_ATTEMPT old = *(struct ETCP_TX_ATTEMPT*)a->attempts->head;
+    etcp_tx_ack(a, old.seq, old.attempt_id, UINT32_MAX);
+    CHECK(etcp_loadbalancer_select_link(conn, 1, NULL) == a);
+    CHECK(conn->active_attempts == 2 * MAX_INFLIGHT_SIZE - 1);
+    for (unsigned i = 0; i < MAX_INFLIGHT_SIZE; i++) CHECK(etcp_tx_queue_ack(a, i+1, i+1) == 0);
+    CHECK(etcp_tx_queue_ack(a, MAX_INFLIGHT_SIZE+1, MAX_INFLIGHT_SIZE+1) < 0);
+    CHECK(a->ack_q->count == MAX_INFLIGHT_SIZE);
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Attempt limits: physical=%u path=%d ACK=%d ready_seq=%u",
+        conn->active_attempts, a->attempts->count, a->ack_q->count, pending->seq);
+    teardown();
+}
+
 /* Отправитель не превращает pending transport в UP до первого INIT/подтверждения сессии. */
 static void initial_readiness(void) {
     setup(); conn->state = 0; conn->initialized = conn->links_up = conn->tx_state = 0;
@@ -639,7 +672,7 @@ int main(void) {
         grow_and_session, allocation_and_send_failure, removed_link_and_bbr, pacing_reset_close, initial_readiness,
         config_limits, timer_failure, immediate_send, ack_without_data_pacing, bbr_delivery_clock, bbr_probe_window, tcp_initial_window,
         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};
+        receive_admission_and_path_acks, peer_window_queries, pacing_continuity, pacing_data_units, attempt_memory_limits};
     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);