Preserve bounded pacing credit across millisecond timer rounding

src/transport_layer/etcp_loadbalancer.c

@@ -33,9 +33,13 @@ void etcp_loadbalancer_set_pacing_rate(struct ETCP_LINK* link, uint32_t rate) {
     if (!link || !link->etcp) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pacing: invalid link=%p", (void*)link); return; }
     uint32_t old_rate = link->bbr_pacing_rate;
     if (rate == old_rate) return;
-    uint64_t now = get_time_tb(), debt = 0;
+    uint64_t now = get_time_tb(), debt = 0, credit = 0, fraction = 0;
     int was_blocked = link->shaper_timer != NULL;
-    if (old_rate && rate && link->shaper_load_time_tb >= now) {
+    if (old_rate && rate && link->shaper_load_time_tb < now) {
+        credit = now - link->shaper_load_time_tb;
+        if (credit > SHAPER_BURST_DELAY_TB) credit = SHAPER_BURST_DELAY_TB;
+        else fraction = link->shaper_sub_nanotime;
+    } else if (old_rate && rate) {
         debt = (link->shaper_load_time_tb - now) * SUB_MODULO + link->shaper_sub_nanotime;
         // debt*old_rate/rate без переполнения: остаток и обе ставки не шире uint32.
         uint64_t quotient = debt / rate, product = (debt % rate) * old_rate;
@@ -48,8 +52,8 @@ void etcp_loadbalancer_set_pacing_rate(struct ETCP_LINK* link, uint32_t rate) {
     }
     link->bbr_pacing_rate = rate;
     link->bandwidth = (uint32_t)(((uint64_t)rate * 8 + 999) / 1000);
-    link->shaper_load_time_tb = now + debt / SUB_MODULO;
-    link->shaper_sub_nanotime = debt % SUB_MODULO;
+    link->shaper_load_time_tb = now + debt / SUB_MODULO - credit;
+    link->shaper_sub_nanotime = credit ? fraction : debt % SUB_MODULO;
     DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] pacing link=%u rate_Bps=%u->%u debt_sub=%llu deadline_tb=%llu now_tb=%llu",
         link->etcp->log_name, link->local_link_id, old_rate, rate, (unsigned long long)debt,
         (unsigned long long)link->shaper_load_time_tb, (unsigned long long)now);
@@ -122,8 +126,8 @@ int etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) {
         // Don't return here, let the function continue to free dgram at the end
     } else {
         uint64_t now_tb = get_time_tb();
-        if (link->shaper_load_time_tb < now_tb) {
-            link->shaper_load_time_tb = now_tb; link->shaper_sub_nanotime = 0;
+        if (link->shaper_load_time_tb + SHAPER_BURST_DELAY_TB < now_tb) {
+            link->shaper_load_time_tb = now_tb - SHAPER_BURST_DELAY_TB; link->shaper_sub_nanotime = 0;
         }
         // Ставка B/s; SUB_MODULO соответствует 0.1мс, один sub равен 0.1нс.
         uint64_t tx_sub = ((uint64_t)pkt_size * 10000000000ULL + link->bbr_pacing_rate - 1) / link->bbr_pacing_rate;

tests/test_etcp_multipath.c

@@ -231,11 +231,7 @@ static void snapshot(const char* phase) {
         struct ETCP_CONN* conn = ep[side].conn;
         for (struct ETCP_LINK* l = conn->links; l; l = l->next) {
             uint64_t occupied = 0; unsigned packets = 0;
-            for (struct ll_entry* e = conn->input_wait_ack->head; e; e = e->next) {
-                struct INFLIGHT_PACKET* pkt = (struct INFLIGHT_PACKET*)e;
-                CHECK(pkt->state == INFLIGHT_STATE_WAIT_ACK);
-                if (pkt->last_link == l) { occupied += pkt->ll.len; packets++; }
-            }
+            for (struct ll_entry* e = l->attempts->head; e; e = e->next) { occupied += e->len; packets++; }
             CHECK(l->inflight_bytes == occupied && l->inflight_packets == packets);
             int p = -1;
             for (unsigned j = 0; j < path_count; j++)

tests/test_etcp_windows.c

@@ -356,6 +356,19 @@ static void pacing_timer_failure(void) {
     etcp_conn_ref_free(conn); teardown();
 }
 
+/* Миллисекундное округление таймеров не должно превращать 1MB/s в 0.5MB/s. */
+static void pacing_continuity(void) {
+    setup(); b->link_status = 0; a->data_rtt_us = 1000000;
+    etcp_loadbalancer_set_pacing_rate(a, 1000000);
+    uint64_t started = get_time_tb();
+    for (unsigned i = 0; i < 50; i++) enqueue(1000);
+    wait_sends(50);
+    uint64_t elapsed = get_time_tb() - started;
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Pacing continuity: DATA=50000 rate=1000000 elapsed_tb=%llu", (unsigned long long)elapsed);
+    CHECK(elapsed >= 450 && elapsed <= 650);
+    ack_all(); teardown();
+}
+
 /* Отправитель не превращает pending transport в UP до первого INIT/подтверждения сессии. */
 static void initial_readiness(void) {
     setup(); conn->state = 0; conn->initialized = conn->links_up = conn->tx_state = 0;
@@ -608,7 +621,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};
+        receive_admission_and_path_acks, peer_window_queries, pacing_continuity};
     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);