Cover initial UP, pacing and bidirectional tight windows

tests/test_etcp_windows.c

@@ -221,11 +221,11 @@ static void removed_link_and_bbr(void) {
     ack(1); CHECK(!conn->unacked_bytes);
     b->bbr = u_calloc(1, sizeof(*b->bbr)); CHECK(b->bbr); bbr_init(b->bbr);
     b->inflight_lim_bytes = 8192;
-    for (int i = 0; i < 8; i++) enqueue(1024);
-    for (int i = 0; i < 8; i++) {
+    for (int i = 0; i < 7; i++) enqueue(1024);
+    for (int i = 0; i < 7; i++) {
         struct ETCP_DGRAM* pkt = request(); CHECK(pkt); memory_pool_free(inst.pkt_pool, pkt);
     }
-    enqueue(100); CHECK(!request() && b->cwnd_limited);
+    enqueue(1025); CHECK(!request() && b->cwnd_limited && b->inflight_bytes < b->inflight_lim_bytes);
     // Первый ACK относится к отправке при наличии ожидающих данных: она не app_limited.
     struct INFLIGHT_PACKET* first = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head;
     CHECK(!first->is_app_limited);
@@ -233,6 +233,39 @@ static void removed_link_and_bbr(void) {
     teardown();
 }
 
+static void shaper_ready(void* arg) {
+    struct ETCP_LINK* link = arg;
+    link->shaper_timer = NULL;
+    loadbalancer_link_ready(link);
+}
+
+static void pacing_reset_close(void) {
+    setup(); b->link_status = 0;
+    a->shaper_timer = uasync_set_timeout(ua, 20, a, shaper_ready, "test_shaper"); CHECK(a->shaper_timer);
+    enqueue(100); CHECK(!request());
+    wait_sends(1); ack_all();
+    enqueue(100); CHECK(conn->tx_timer);
+    etcp_conn_reinit_id(conn, "queued sender reset", conn->reset_id);
+    CHECK(!conn->tx_timer && !conn->unacked_bytes && conn->reinit_pending);
+    enqueue(100); CHECK(conn->tx_timer);
+    CHECK(etcp_conn_ref_take(conn) == 0);
+    etcp_connection_close(conn); CHECK(conn->state == 2 && !conn->tx_timer);
+    etcp_conn_ref_free(conn);
+    teardown();
+}
+
+/* Отправитель не превращает pending transport в UP до первого INIT/подтверждения сессии. */
+static void initial_readiness(void) {
+    setup(); conn->state = 0; conn->initialized = conn->links_up = conn->tx_state = 0;
+    conn->session_required = 1; enqueue(100);
+    test_clock_run_for(ua, 20);
+    CHECK(!conn->tx_state && !sends && !conn->links_up);
+    conn->link_ready_for_send_fn(conn); CHECK(!conn->links_up);
+    conn->session_required = 0; etcp_conn_ready(conn); wait_sends(1);
+    CHECK(conn->state == 1 && conn->links_up && conn->initialized);
+    ack_all(); teardown();
+}
+
 static void config_limits(void) {
     const char* values[] = {"0", "8191", "-1", "2147483648", "999999999999999999999", "abc", "8192x", "8192", "65536"};
     for (unsigned i = 0; i < sizeof(values) / sizeof(values[0]); i++) {
@@ -249,7 +282,7 @@ int main(void) {
     test_clock_init(10);
     size_t baseline = u_get_allocated_count();
     void (*cases[])(void) = {selection, boundaries_ack_burst, late_ack_and_migration, lost_ack_and_shrink,
-        grow_and_session, allocation_and_send_failure, removed_link_and_bbr, config_limits};
+        grow_and_session, allocation_and_send_failure, removed_link_and_bbr, pacing_reset_close, initial_readiness, config_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);

tests/test_stcp_traffic.c

@@ -28,8 +28,7 @@ static int test_failed = 0;
 } while(0)
 
 struct rx_ctx {
-    uint8_t *buf;
-    size_t len, cap;
+    size_t len;
     int pkts;
     int done;
     size_t expected_total;
@@ -41,32 +40,22 @@ static struct ETCP_CONN *srv_conn, *cli_conn;
 static struct NODE_CONN_DIRECT *cli_owner;
 static int srv_send_ready, cli_send_ready;
 
-static void srv_recv_cb(struct ETCP_CONN *conn, struct ll_entry *entry) {
-    (void)conn;
-    if (!entry->dgram || entry->len < 5) { queue_dgram_free(entry); queue_entry_free(entry); return; }
-    srv_rx.pkts++;
-    size_t pay = entry->len - 5;
-    size_t need = srv_rx.len + pay;
-    if (need > srv_rx.cap) { srv_rx.cap = need + 65536; srv_rx.buf = u_realloc(srv_rx.buf, srv_rx.cap); }
-    memcpy(srv_rx.buf + srv_rx.len, entry->dgram + 5, pay);
-    srv_rx.len += pay;
-    queue_dgram_free(entry);
-    queue_entry_free(entry);
-    if (srv_rx.len >= srv_rx.expected_total) srv_rx.done = 1;
-}
-
-static void cli_recv_cb(struct ETCP_CONN *conn, struct ll_entry *entry) {
-    (void)conn;
-    if (!entry->dgram || entry->len < 5) { queue_dgram_free(entry); queue_entry_free(entry); return; }
-    cli_rx.pkts++;
-    size_t pay = entry->len - 5;
-    size_t need = cli_rx.len + pay;
-    if (need > cli_rx.cap) { cli_rx.cap = need + 65536; cli_rx.buf = u_realloc(cli_rx.buf, cli_rx.cap); }
-    memcpy(cli_rx.buf + cli_rx.len, entry->dgram + 5, pay);
-    cli_rx.len += pay;
-    queue_dgram_free(entry);
-    queue_entry_free(entry);
-    if (cli_rx.len >= cli_rx.expected_total) cli_rx.done = 1;
+/* Проверяем порядок, отсутствие дублей и каждый байт, включая трафик сразу после NCD UP. */
+static void receive_packet(struct ETCP_CONN* conn, struct ll_entry* entry) {
+    struct rx_ctx* rx = conn->instance == srv_inst ? &srv_rx : &cli_rx;
+    unsigned side = rx == &srv_rx ? 1 : 2, seq = 0;
+    int valid = entry->dgram && entry->len >= 5;
+    if (valid) {
+        for (unsigned i = 0; i < 4; i++) seq |= (unsigned)entry->dgram[i + 1] << (8 * i);
+        valid = entry->dgram[0] == side && seq == (unsigned)rx->pkts;
+        for (unsigned i = 5; valid && i < entry->len; i++)
+            valid = entry->dgram[i] == (uint8_t)(i * 7 + seq * 13 + side);
+    }
+    if (!valid) {
+        DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Invalid STCP payload side=%u seq=%u expected=%d len=%u", side, seq, rx->pkts, entry->len);
+        test_failed = 1;
+    } else { rx->pkts++; rx->len += entry->len - 5; rx->done = rx->len == rx->expected_total; }
+    queue_dgram_free(entry); queue_entry_free(entry);
 }
 
 static void connect_cb(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg) {
@@ -101,7 +90,9 @@ static int send_etcp_packet(struct ETCP_CONN *conn, int side, int seq, size_t pa
     if (!e->dgram) { queue_entry_free(e); return -1; }
     memcpy(e->dgram, buf, tot);
     e->len = (uint16_t)tot;
-    return etcp_send(conn, e);
+    int rc = etcp_send(conn, e);
+    if (rc < 0) { queue_dgram_free(e); queue_entry_free(e); }
+    return rc;
 }
 
 static char *build_server_config(int port) {
@@ -109,6 +100,7 @@ static char *build_server_config(int port) {
     snprintf(buf, sizeof(buf),
         "[global]\n"
         "my_node_id=0xAAAAAAAAAAAAAAAA\n"
+        "bbr_max_cwnd=8192\n"
         "my_private_key=38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68\n"
         "my_public_key=ce8871f07fa056c636d297115f231b08c29cdf94e0d440fce83a07c34416d36a\n"
         "\n"
@@ -125,6 +117,7 @@ static char *build_client_config(int port) {
     snprintf(buf, sizeof(buf),
         "[global]\n"
         "my_node_id=0xBBBBBBBBBBBBBBBB\n"
+        "bbr_max_cwnd=8192\n"
         "my_private_key=704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f\n"
         "my_public_key=b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01\n"
         "\n"
@@ -156,8 +149,8 @@ int main(void) {
 
     memset(&srv_rx, 0, sizeof(srv_rx)); srv_rx.expected_total = TRAFFIC_MB;
     memset(&cli_rx, 0, sizeof(cli_rx)); cli_rx.expected_total = TRAFFIC_MB;
-    TASSERT(etcp_bind(srv_inst, 1, srv_recv_cb) >= 0);
-    TASSERT(etcp_bind(cli_inst, 2, cli_recv_cb) >= 0);
+    TASSERT(etcp_bind(srv_inst, 1, receive_packet) >= 0);
+    TASSERT(etcp_bind(cli_inst, 2, receive_packet) >= 0);
 
     struct TOPO_GROUP_NODE *srv_node = topo_groups_get_default(srv_inst->topo_groups) ? topo_groups_get_default(srv_inst->topo_groups)->local_node : NULL;
     TASSERT(srv_node);
@@ -197,14 +190,14 @@ int main(void) {
         if (!srv_done && remaining_s > 0) {
             size_t sz = PKT_MIN + (rand() % (PKT_MAX - PKT_MIN + 1));
             if (sz > remaining_s) sz = remaining_s;
-            if (send_etcp_packet(srv_conn, 2, seq_s++, sz) == 0) remaining_s -= sz;
+            if (send_etcp_packet(srv_conn, 2, seq_s, sz) == 0) { remaining_s -= sz; seq_s++; }
         }
         if (remaining_s == 0 && seq_s > 0) srv_done = 1;
 
         if (!cli_done && remaining_c > 0) {
             size_t sz = PKT_MIN + (rand() % (PKT_MAX - PKT_MIN + 1));
             if (sz > remaining_c) sz = remaining_c;
-            if (send_etcp_packet(cli_conn, 1, seq_c++, sz) == 0) remaining_c -= sz;
+            if (send_etcp_packet(cli_conn, 1, seq_c, sz) == 0) { remaining_c -= sz; seq_c++; }
         }
         if (remaining_c == 0 && seq_c > 0) cli_done = 1;
 
@@ -217,11 +210,7 @@ int main(void) {
     DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "srv_rx: %zu bytes, %d pkts, done=%d", srv_rx.len, srv_rx.pkts, srv_rx.done);
     DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "cli_rx: %zu bytes, %d pkts, done=%d", cli_rx.len, cli_rx.pkts, cli_rx.done);
 
-    TASSERT(srv_rx.len >= TRAFFIC_MB);
-    TASSERT(cli_rx.len >= TRAFFIC_MB);
-
-    if (srv_rx.buf) u_free(srv_rx.buf);
-    if (cli_rx.buf) u_free(cli_rx.buf);
+    TASSERT(!test_failed && srv_rx.len == TRAFFIC_MB && cli_rx.len == TRAFFIC_MB);
 
     node_conn_direct_close(cli_owner); cli_owner = NULL;
     srv_inst->running = 0; cli_inst->running = 0;