Bound ETCP bandwidth samples by receiver delivery progress

src/transport_layer/etcp_connections.h

@@ -84,6 +84,12 @@ struct ETCP_DGRAM {// пакет (незашифрованный)
 
 // Размер базового ACK: type(1) + count(1) + last_delivered_id(4) + rx_dup_count(2)
 #define ETCP_ACK_BASE_SIZE (2 + 4 + 2)
+#define ETCP_ACK_RX_POINT_SIZE 16 // При count>0 после entries: принятые DATA-байты u64 и время их приёма u64 (LE, мкс)
+
+// Размер ACK-секции для count=0..255; пустой ACK не содержит точки приёма и остаётся 8-байтовым.
+static inline unsigned etcp_ack_section_size(uint8_t count) {
+    return ETCP_ACK_BASE_SIZE + (unsigned)count * ETCP_ACK_ENTRY_SIZE + (count ? ETCP_ACK_RX_POINT_SIZE : 0);
+}
 
 // INIT внутри шифрованного тела, до padding: запрос — 67 байт, ответ — 64 байта.
 // Каждая сторона передаёт свою эпоху reset_id; смена известной эпохи подтверждается через etcp_session.
@@ -341,6 +347,10 @@ struct ETCP_LINK {
     uint64_t    delivered_bytes;        // кумулятивно delivered bytes на линке (для round detection)
     uint64_t    last_ack_time_us;       // время последней доставки/начала первого flight в микросекундах
     uint64_t    first_tx_time_us;       // начало интервала отправок, продвигается подтверждёнными попытками
+    uint64_t    rx_data_bytes;          // допустимые физические DATA-байты пути, включая копии; сбрасывается с потоком
+    uint64_t    rx_data_time_us;        // время последнего увеличения rx_data_bytes, локальные монотонные мкс; 0 — точки нет
+    uint64_t    peer_rx_data_bytes;     // последняя достоверная точка из ACK действующей попытки этого пути
+    uint64_t    peer_rx_data_time_us;   // часы получателя, не сравниваются с локальными; 0 — нет базы текущего flight
     uint32_t    bbr_pacing_rate;        // Точный pacing B/s; менять через set_pacing_rate для пересчёта долга/таймера
     uint32_t    bbr_loss_since_ack;     // счётчик потерь на линке с последнего ACK (для rs->lost)
 

src/transport_layer/etcp_tx.c

@@ -100,6 +100,8 @@ static void retransmit_cb(void* arg) {
 }
 
 int etcp_tx_link_init(struct ETCP_LINK* link) {
+    link->rx_data_bytes = link->rx_data_time_us = 0;
+    link->peer_rx_data_bytes = link->peer_rx_data_time_us = 0;
     link->attempts = queue_new(link->etcp->instance->ua, 128, 0, 8, "ETCP attempts");
     link->ack_q = queue_new(link->etcp->instance->ua, 128, 0, 8, "ETCP path ACK");
     if (link->attempts && link->ack_q) {
@@ -129,6 +131,8 @@ void etcp_tx_link_reset(struct ETCP_LINK* link, int retry) {
     link->data_rtt_us = link->data_jitter_us = link->bbr_loss_since_ack = 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;
+    link->peer_rx_data_bytes = link->peer_rx_data_time_us = 0;
     link->last_ack_time_us = get_time_us(); link->first_tx_time_us = 0;
     link->cwnd_limited = 0;
     if (link->bbr && link->etcp->normalizer) {
@@ -184,10 +188,15 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) {
         abort();
     }
     uint64_t now = get_time_us();
-    if (link->inflight_packets == 1) { link->first_tx_time_us = now; link->last_ack_time_us = now; }
+    if (link->inflight_packets == 1) {
+        link->first_tx_time_us = now; link->last_ack_time_us = now;
+        link->peer_rx_data_bytes = link->peer_rx_data_time_us = 0;
+    }
     attempt->sent_at_us = now; attempt->delivered_at_send = link->delivered_bytes;
     attempt->delivered_time_us = link->last_ack_time_us; attempt->first_tx_us = link->first_tx_time_us;
     attempt->inflight_at_send = link->inflight_bytes;
+    attempt->peer_rx_bytes_at_send = link->peer_rx_data_bytes;
+    attempt->peer_rx_time_at_send = link->peer_rx_data_time_us;
     struct PKTNORM* pn = conn->normalizer;
     attempt->is_app_limited = !conn->input_queue->count && conn->input_send_q->count == 1 &&
         (!pn || (!pn->input->count && !pn->pending && !pn->data_ptr));
@@ -207,7 +216,8 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) {
     retransmit_arm(link);
 }
 
-void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us) {
+void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us,
+                 uint64_t peer_rx_bytes, uint64_t peer_rx_time_us) {
     struct ETCP_CONN* conn = link->etcp;
     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)) {
@@ -217,6 +227,20 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint
     }
     if (attempt) {
         struct ETCP_TX_ATTEMPT sample = *attempt;
+        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) ||
+             (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,
+                (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;
+        }
+        if (rx_valid && peer_rx_bytes > link->peer_rx_data_bytes) {
+            link->peer_rx_data_bytes = peer_rx_bytes; link->peer_rx_data_time_us = peer_rx_time_us;
+        }
         int limited = link->cwnd_limited || link->inflight_bytes >= link->inflight_lim_bytes;
         attempt_retire(link, attempt, 0, 0);
         uint64_t now = get_time_us(), elapsed = now - sample.sent_at_us;
@@ -237,10 +261,28 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint
         uint64_t interval = ack_elapsed > send_elapsed ? ack_elapsed : send_elapsed;
         struct bbr_rate_sample rs = {
             .delivered = delivered > UINT32_MAX ? UINT32_MAX : (uint32_t)delivered,
-            .interval_us = interval > UINT32_MAX ? UINT32_MAX : (uint32_t)interval,
+            .interval_us = interval > UINT32_MAX ? 0 : (uint32_t)interval,
             .rtt_us = rtt, .acked_sacked = sample.ll.len, .prior_delivered = (uint32_t)sample.delivered_at_send,
             .tx_in_flight = sample.inflight_at_send, .lost = (int)link->bbr_loss_since_ack, .is_app_limited = sample.is_app_limited
         };
+        uint64_t rx_bytes = 0, rx_us = 0;
+        if (rs.interval_us && rx_valid && sample.peer_rx_time_at_send &&
+            link->peer_rx_data_bytes > sample.peer_rx_bytes_at_send &&
+            link->peer_rx_data_time_us > sample.peer_rx_time_at_send) {
+            rx_bytes = link->peer_rx_data_bytes - sample.peer_rx_bytes_at_send;
+            rx_us = link->peer_rx_data_time_us - sample.peer_rx_time_at_send;
+            if (rx_us > UINT32_MAX) rs.interval_us = 0;
+            else {
+                uint64_t numerator = (uint64_t)rs.delivered * rx_us;
+                uint64_t bound = numerator / rx_bytes + !!(numerator % rx_bytes);
+                if (bound > UINT32_MAX) rs.interval_us = 0;
+                else if (bound > rs.interval_us) rs.interval_us = (uint32_t)bound;
+            }
+        }
+        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,
+                (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;
         if (link->bbr) {
@@ -249,8 +291,10 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint
             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 "
-            "flight=%u cwnd=%u pace=%u RTO_us=%u", conn->log_name, seq, (unsigned long long)attempt_id, link->local_link_id,
-            rtt, receiver_delay_us, rs.delivered, rs.interval_us, link->inflight_bytes, link->inflight_lim_bytes,
+            "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,
+            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;

src/transport_layer/etcp_tx.h

@@ -23,6 +23,8 @@ struct ETCP_TX_ATTEMPT {
     uint64_t delivered_at_send;     // доставленные DATA-байты этого канала на момент отправки
     uint64_t delivered_time_us;     // начало ACK-интервала оценки скорости
     uint64_t first_tx_us;           // начало интервала отправки
+    uint64_t peer_rx_bytes_at_send; // точка приёма, известная при отправке этой попытки
+    uint64_t peer_rx_time_at_send;  // время этой точки на часах получателя; 0 — новая/неизвестная база
     uint32_t inflight_at_send;      // окно после включения этой попытки
     uint8_t is_app_limited;         // источник не имел ожидающих данных либо идёт PROBE_RTT
 };
@@ -44,8 +46,12 @@ struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_
  * Вызвать ровно один раз до освобождения dgram и до следующего poll; dgram->attempt обнуляется. */
 void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success);
 /* Принятый ACK: seq подтверждает логическую доставку; attempt_id обновляет окно/BBR только своего линка.
- * Неизвестный старый ID не создаёт измерение и не вычитает окно. UINT32_MAX delay — RTT недостоверен. */
-void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us);
+ * Неизвестный старый ID не создаёт измерение и не вычитает окно. UINT32_MAX delay — RTT недостоверен.
+ * peer_rx_bytes/time_us — согласованная точка приёма пути из ACK; 0/0 — неизвестна.
+ * Достоверная пара ограничивает BW темпом приёма, сохраняя границы send/ACK; RTT и доставка от неё не зависят.
+ * Непредставимый интервал не создаёт оценку BW. Пустой flight заново начинает приёмную базу, исключая простой. */
+void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us,
+                 uint64_t peer_rx_bytes, uint64_t peer_rx_time_us);
 /* Поставить ACK уже сохранённого/доставленного DATA в очередь этого линка; 0/-1, ошибка логируется. */
 int etcp_tx_queue_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id);