Exercise NAT backpressure, mapping reuse, copy ownership and shutdown

tests/Makefile.am

@@ -528,7 +528,7 @@ test_nat_transport_SOURCES = test_nat_transport.c
 test_nat_transport_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
 test_nat_transport_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
 if OS_LINUX
-test_nat_transport_LDFLAGS = -Wl,--wrap=etcp_route_send
+test_nat_transport_LDFLAGS = -Wl,--wrap=etcp_route_send -Wl,--wrap=queue_data_put -Wl,--wrap=tun_write
 endif
 
 test_nat_stress_SOURCES = test_nat_stress.c

tests/test_nat_transport.c

@@ -49,6 +49,18 @@ static int send_test_mode, send_test_calls, send_test_valid;
 static struct UTUN_INSTANCE* send_test_instance;
 static uint64_t send_test_destination;
 static size_t send_test_ip_len;
+static struct ll_entry* original_tun_packet;
+static struct tun_if* write_blocked_tun;
+int __real_queue_data_put(struct ll_queue*, struct ll_entry*);
+int __wrap_queue_data_put(struct ll_queue* q, struct ll_entry* entry) {
+    if (send_test_mode && send_test_instance && q == send_test_instance->nat_tr.nat_tun->output_queue) original_tun_packet = entry;
+    return __real_queue_data_put(q, entry);
+}
+ssize_t __real_tun_write(struct tun_if*, const uint8_t*, size_t);
+ssize_t __wrap_tun_write(struct tun_if* tun, const uint8_t* data, size_t len) {
+    if (tun == write_blocked_tun) { errno = EAGAIN; return -1; }
+    return __real_tun_write(tun, data, len);
+}
 
 int __real_etcp_route_send(struct UTUN_INSTANCE*, uint64_t, uint64_t, struct ll_entry*, int, int);
 int __wrap_etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t destination,
@@ -56,7 +68,8 @@ int __wrap_etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t
     if (!send_test_mode) return __real_etcp_route_send(inst, group, destination, entry, force, mode);
     send_test_calls++;
     if (inst != send_test_instance || group != TOPO_GROUP_UTUN || destination != send_test_destination ||
-        !entry || !entry->dgram || entry->len != send_test_ip_len + 1 || entry->dgram[0] != ETCP_RT_ID_NAT || force || mode) {
+        !entry || entry != original_tun_packet || !entry->dgram || entry->len != send_test_ip_len + 1 ||
+        entry->dgram[0] != ETCP_RT_ID_NAT || force || mode) {
         DEBUG_ERROR(DEBUG_CATEGORY_NAT, "Unexpected NAT send in ownership test: destination=%016llx call=%d",
                     (unsigned long long)destination, send_test_calls);
         send_test_valid = 0;
@@ -600,6 +613,123 @@ static int test_full_roundtrip(void) {
     return 1;
 }
 
+/* Заполнить router send_q без потерь; затем проверить порядок. У провайдера mapping переиспользуется во время ожидания. */
+static int test_router_backpressure(int provider) {
+    struct UTUN_INSTANCE* sender = provider ? inst_provider : inst_client;
+    struct tun_if* sink = provider ? inst_client->nat_tr.nat_tun : inst_provider->nat_tr.nat_tun;
+    uint64_t destination = provider ? g_client_node_id : g_provider_node_id;
+    struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(sender, TOPO_GROUP_UTUN, destination, ETCP_RT_ID_NAT);
+    if (!route || queue_entry_count(route->send_q)) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "Backpressure fixture requires empty router queue"); return 0; }
+    size_t len;
+    uint8_t* packet;
+    uint16_t external_port = 0;
+    if (provider) {
+        packet = build_udp_pkt(0x0a0000dd, 45678, 0x08080808, 4242, &len);
+        if (eim_nat_egress(&inst_provider->nat, packet, len, g_client_node_id, NULL)) { free(packet); return 0; }
+        memcpy(&external_port, packet + 20, 2); external_port = be16toh(external_port);
+        free(packet);
+        packet = build_udp_pkt(0x08080808, 4242, inst_provider->nat.gateway_ip, external_port, &len);
+    } else packet = build_udp_pkt(0x0a0000ee, 40001, 0x08080808, 4242, &len);
+    route->peer_receive_blocked = 1;
+    uint64_t dropped = sender->nat_tr.dropped;
+    int ok = 1;
+    for (uint32_t i = 0; i < 66; i++) {
+        memcpy(packet + 28, &i, 4);
+        if (tun_inject_packet(sender->nat_tr.nat_tun, packet, len)) ok = 0;
+    }
+    free(packet);
+    if (queue_entry_count(route->send_q) != ROUTER_MAX_SEND_Q_PACKETS || sender->nat_tr.pending_count != 2 ||
+        sender->nat_tr.dropped != dropped) ok = 0;
+    if (provider) {
+        if (eim_nat_remove(&inst_provider->nat, external_port) != 1 ||
+            eim_nat_add_forward(&inst_provider->nat, IPPROTO_UDP_UINT8, 0x0a000077, htobe16(55000), external_port, 0x12345678)) ok = 0;
+    }
+    route->peer_receive_blocked = 0;
+    uint16_t previous_limit = route->max_inflight;
+    etcp_router_set_max_inflight(route, previous_limit + 1); // Возобновить drain без добавления фиктивного IP-пакета.
+    uint32_t received = 0;
+    for (int cycles = 0; cycles < 1000 && !test_timed_out && received < 66; cycles++) {
+        uasync_poll(ua, 5);
+        uint8_t data[1500]; ssize_t n;
+        while ((n = tun_read_packet(sink, data, sizeof(data))) > 0) {
+            uint16_t port = 0;
+            if (n >= 28) memcpy(&port, data + 22, 2);
+            if (port != htobe16(provider ? 45678 : 4242)) continue;
+            uint32_t serial; memcpy(&serial, data + 28, 4);
+            if (n != (ssize_t)len || serial != received || !verify_ip_checksum(data)) ok = 0;
+            received++;
+        }
+    }
+    etcp_router_set_max_inflight(route, previous_limit);
+    if (received != 66 || sender->nat_tr.pending_count || sender->nat_tr.dropped != dropped) ok = 0;
+    if (provider) eim_nat_remove(&inst_provider->nat, external_port);
+    if (ok) DEBUG_INFO(DEBUG_CATEGORY_NAT, "PASS %s router backpressure: 66/66 ordered, no drops, mapping reuse safe", provider ? "provider" : "client");
+    else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL %s router backpressure: received=%u queued=%d pending=%zu dropped=%llu",
+                     provider ? "provider" : "client", received, queue_entry_count(route->send_q), sender->nat_tr.pending_count,
+                     (unsigned long long)(sender->nat_tr.dropped - dropped));
+    return ok;
+}
+
+#ifdef __linux__
+/* EAGAIN удерживает первый IP-пакет; второй остаётся в router recv_q. После возобновления оба доставляются по порядку. */
+static int test_tun_write_backpressure(void) {
+    write_blocked_tun = inst_client->nat_tr.nat_tun;
+    for (uint32_t i = 1; i <= 2; i++) {
+        size_t len;
+        uint8_t* ip = build_udp_pkt(0x08080808, 53, 0x0a0000cd, 44444, &len);
+        memcpy(ip + 28, &i, 4);
+        struct ll_entry* packet = ll_alloc_lldgram(len + 1);
+        if (!packet) { free(ip); write_blocked_tun = NULL; return 0; }
+        packet->len = len + 1; packet->dgram[0] = ETCP_RT_ID_NAT;
+        memcpy(packet->dgram + 1, ip, len); free(ip);
+        if (etcp_route_send(inst_provider, TOPO_GROUP_UTUN, g_client_node_id, packet, 0, 0)) { write_blocked_tun = NULL; return 0; }
+    }
+    struct ETCP_ROUTER_CONN* incoming = etcp_router_conn_get(inst_client, TOPO_GROUP_UTUN, g_provider_node_id, ETCP_RT_ID_NAT);
+    for (int cycles = 0; cycles < 500 && !test_timed_out && !incoming->receive_blocked; cycles++) uasync_poll(ua, 5);
+    int ok = inst_client->nat_tr.tun_pending && incoming->receive_blocked;
+    write_blocked_tun = NULL;
+    uint32_t received = 0;
+    for (int cycles = 0; cycles < 500 && !test_timed_out && received < 2; cycles++) {
+        uasync_poll(ua, 5);
+        uint8_t data[1500]; ssize_t n;
+        while ((n = tun_read_packet(inst_client->nat_tr.nat_tun, data, sizeof(data))) > 0) {
+            uint32_t serial = 0;
+            if (n >= 32) memcpy(&serial, data + 28, 4);
+            if (n != 42 || serial != received + 1) ok = 0;
+            received++;
+        }
+    }
+    if (received != 2 || inst_client->nat_tr.tun_pending || incoming->receive_blocked) ok = 0;
+    if (ok) DEBUG_INFO(DEBUG_CATEGORY_NAT, "PASS TUN write backpressure: retained packet, router pause and ordered resume");
+    else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL TUN write backpressure: received=%u pending=%p blocked=%u", received,
+                     (void*)inst_client->nat_tr.tun_pending, incoming->receive_blocked);
+    return ok;
+}
+#endif
+
+/* Один недоступный адресат исчерпывает только свой буфер. Stop отменяет pending/waiters до уничтожения TUN. */
+static int test_slow_peer_and_shutdown(void) {
+    const uint64_t slow_node = UINT64_C(0xfedcba9876543210);
+    if (eim_nat_add_forward(&inst_provider->nat, IPPROTO_UDP_UINT8, 0x0a000099, htobe16(9091), 9090, slow_node)) return 0;
+    size_t len;
+    uint8_t* packet = build_udp_pkt(0x08080808, 53, inst_provider->nat.gateway_ip, 9090, &len);
+    uint64_t before = inst_provider->nat_tr.dropped;
+    for (int i = 0; i < ROUTER_MAX_SEND_Q_PACKETS + NAT_PENDING_PER_PEER + 1; i++)
+        if (tun_inject_packet(inst_provider->nat_tr.nat_tun, packet, len)) { free(packet); return 0; }
+    free(packet);
+    int ok = inst_provider->nat_tr.pending_count == NAT_PENDING_PER_PEER && inst_provider->nat_tr.dropped == before + 1;
+    if (!test_static_forward_delivery()) ok = 0; // Другой адресат продолжает получать пакеты.
+    nat_transport_destroy(inst_provider);
+    uasync_poll(ua, 0);
+    if (inst_provider->nat_tr.initialized || inst_provider->nat_tr.pending_count || inst_provider->nat.initialized) ok = 0;
+    for (struct ll_entry* e = inst_provider->router_conns->head; e; e = e->next)
+        if (((struct ETCP_ROUTER_CONN*)e)->svc_id == ETCP_RT_ID_NAT) ok = 0;
+    if (nat_transport_init(inst_provider)) ok = 0;
+    if (ok) DEBUG_INFO(DEBUG_CATEGORY_NAT, "PASS slow-peer isolation and stop/restart with pending TUN-pool packets");
+    else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL slow-peer isolation/shutdown");
+    return ok;
+}
+
 // ==================== Main ====================
 int main(void) {
     int test_result = 1;
@@ -673,6 +803,12 @@ int main(void) {
     // Реальная доставка после инъекций проверяет восстановление обеих веток без перезапуска NAT.
     total++; if (test_full_roundtrip()) passed++; else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL: full_roundtrip");
     total++; if (test_static_forward_delivery()) passed++; else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL: static_forward");
+    total++; if (test_router_backpressure(0)) passed++; else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL: client_backpressure");
+    total++; if (test_router_backpressure(1)) passed++; else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL: provider_backpressure");
+#ifdef __linux__
+    total++; if (test_tun_write_backpressure()) passed++; else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL: tun_write_backpressure");
+#endif
+    total++; if (test_slow_peer_and_shutdown()) passed++; else DEBUG_ERROR(DEBUG_CATEGORY_NAT, "FAIL: slow_peer_shutdown");
 
     printf("\n=== NAT Transport Tests: %d/%d passed ===\n", passed, total);
     test_result = (passed == total) ? 0 : 1;