From d0c83cfff12f826b711932815262e21ae1f118a7 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Wed, 21 Oct 2026 22:30:01 +0200
Subject: [PATCH] Add bounded NAT backpressure and forward TUN packets without
 copying

---
 src/nat_transport.c | 437 ++++++++++++++++++++++++++------------------
 src/nat_transport.h |  14 ++
 2 files changed, 278 insertions(+), 173 deletions(-)

diff --git a/src/nat_transport.c b/src/nat_transport.c
index 20e5916d..8630f1b8 100644
--- a/src/nat_transport.c
+++ b/src/nat_transport.c
@@ -1,223 +1,314 @@
 #include "nat_transport.h"
 #include "eim_nat.h"
-#include "etcp.h"
 #include "utun_instance.h"
 #include "config_parser.h"
 #include "tun_if.h"
-#include "etcp_api.h"
 #include "etcp_router.h"
-#include "../routing_layer/topo_node.h"
 #include "../lib/debug_config.h"
 #include "../lib/mem.h"
 #include "../lib/ll_queue.h"
+#include "../lib/u_async.h"
 #include <string.h>
+#include <errno.h>
 
-#define NAT_SVC_HDR_SIZE 1   // send: svc_id(1) + ip_data (src/dst добавляет роутер)
+#define NAT_MAINTENANCE_TB 1000
+#define NAT_WRITE_RETRY_TB 100
 
-static ip_str_t ip_host_to_str(uint32_t ip_host) {
-    struct in_addr a; a.s_addr = htonl(ip_host);
-    return ip_to_str(&a, AF_INET);
+// Один владелец FIFO и ожидания на каждого адресата. node_id начинается с ll.data для хеш-индекса.
+struct nat_peer {
+    struct ll_entry ll;
+    uint64_t node_id;
+    struct UTUN_INSTANCE* inst;
+    struct ll_queue* packets;
+    struct ll_entry* pending;
+    struct queue_waiter_handle waiter;
+    uint64_t last_activity;
+};
+
+static void nat_peer_send_ready(struct ll_queue* q, void* arg);
+static void nat_tun_retry(void* arg);
+
+// Освободить пакет с сохранением его allocator/pool metadata.
+static void nat_packet_free(struct ll_entry* packet) {
+    if (packet) { queue_dgram_free(packet); queue_entry_free(packet); }
 }
 
-// ==================== Callbacks ====================
-
-// CLIENT: NAT TUN output → encapsulate in ETCP_RT_ID_NAT → send to provider via etcp_router
-static void nat_transport_client_tun_out_cb(struct ll_queue* q, void* arg) {
-    struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
-    if (!inst) { queue_resume_callback(q); return; }
+// Перегрузка одного получателя не останавливает остальных; диагностировать суммарные потери не чаще раза в секунду.
+static void nat_drop(struct UTUN_INSTANCE* inst, struct ll_entry* packet, uint64_t destination, const char* reason) {
     struct nat_transport_ctx* tr = &inst->nat_tr;
-
-    struct ll_entry* pkt = queue_data_get(q);
-    if (!pkt) { queue_resume_callback(q); return; }
-    if (!pkt->dgram || pkt->len < 2) { queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
-
-    size_t ip_len    = pkt->len - 1;
-    size_t total_len = NAT_SVC_HDR_SIZE + ip_len;
-    uint8_t* new_dgram = u_malloc(total_len);
-    if (!new_dgram) { queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
-
-    new_dgram[0] = ETCP_RT_ID_NAT;
-    memcpy(new_dgram + 1, pkt->dgram + 1, ip_len);
-
-    struct ll_entry* new_entry = queue_entry_new(0);
-    if (!new_entry) { u_free(new_dgram); queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
-    new_entry->dgram = new_dgram;
-    new_entry->len   = total_len;
-
-    queue_dgram_free(pkt); queue_entry_free(pkt);
-    queue_resume_callback(q);
-
-    // Роутер принимает владение new_entry и dgram при любом результате, включая ошибку.
-    int ret = etcp_route_send(inst, TOPO_GROUP_UTUN, tr->nat_via_node_id, new_entry, 0, 0);
-    if (ret != 0) {
-        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT client: etcp_route_send to provider %016llx failed: error=%d ip_bytes=%zu",
-                   (unsigned long long)tr->nat_via_node_id, ret, ip_len);
+    tr->dropped++;
+    uint64_t now = get_time_tb();
+    if (!tr->last_drop_log || now - tr->last_drop_log >= 10000) {
+        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT drop: reason=%s destination=%016llx total=%llu pending=%zu",
+                   reason, (unsigned long long)destination, (unsigned long long)tr->dropped, tr->pending_count);
+        tr->last_drop_log = now;
     }
+    nat_packet_free(packet);
 }
 
-// PROVIDER: NAT TUN output (internet response) → ingress NAT → send back via etcp_router
-static void nat_transport_provider_tun_out_cb(struct ll_queue* q, void* arg) {
-    struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
-    if (!inst) { queue_resume_callback(q); return; }
-    struct eim_nat_ctx* ctx = &inst->nat;
-
-    struct ll_entry* pkt = queue_data_get(q);
-    if (!pkt) { queue_resume_callback(q); return; }
-    if (!pkt->dgram || pkt->len < 2) { queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
-
-    struct eim_nat_entry* entry = NULL;
-    int ret = eim_nat_ingress(ctx, pkt->dgram + 1, pkt->len - 1, &entry);
-    if (ret != 0 || !entry || entry->src_node_id == 0) {
-        if (ret == 0) DEBUG_WARN(DEBUG_CATEGORY_NAT, "Ingress NAT: no matching entry, dropping");
-        queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q);
+// Передать удержанный пакет, когда у роутера есть место; waiter не хранит указатель на NAT mapping.
+static void nat_peer_send_ready(struct ll_queue* q, void* arg) {
+    (void)q;
+    struct nat_peer* peer = arg;
+    struct UTUN_INSTANCE* inst = peer->inst;
+    struct nat_transport_ctx* tr = &inst->nat_tr;
+    if (!peer->pending) return;
+    struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(inst, TOPO_GROUP_UTUN, peer->node_id, ETCP_RT_ID_NAT);
+    if (!route || route->closed) return; // maintenance повторит после завершения close/reset.
+    if (!etcp_router_send_q_has_room(inst, TOPO_GROUP_UTUN, peer->node_id, ETCP_RT_ID_NAT)) {
+        if (!peer->waiter.internal && !peer->waiter.call_soon_id) {
+            DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT waiting for router: destination=%016llx pending=%zu",
+                        (unsigned long long)peer->node_id, tr->pending_count);
+            etcp_router_on_send_ready(inst, TOPO_GROUP_UTUN, peer->node_id, ETCP_RT_ID_NAT,
+                                      &peer->waiter, nat_peer_send_ready, peer);
+        }
         return;
     }
+    struct ll_entry* packet = peer->pending;
+    size_t ip_len = packet->len - 1;
+    peer->pending = NULL;
+    tr->pending_count--;
+    peer->last_activity = get_time_tb();
+    // Роутер синхронно копирует payload и освобождает исходный TUN entry/dgram при любом результате.
+    int result = etcp_route_send(inst, TOPO_GROUP_UTUN, peer->node_id, packet, 0, 0);
+    if (result) {
+        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT router send failed: destination=%016llx error=%d ip_bytes=%zu",
+                   (unsigned long long)peer->node_id, result, ip_len);
+    }
+    queue_resume_callback(peer->packets);
+    if (tr->blocked_packet && tr->pending_count < NAT_PENDING_TOTAL) queue_resume_callback(tr->nat_tun->output_queue);
+}
 
-    size_t ip_len    = pkt->len - 1;
-    size_t total_len = NAT_SVC_HDR_SIZE + ip_len;
-    uint8_t* new_dgram = u_malloc(total_len);
-    if (!new_dgram) { queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
+// Забрать ровно один пакет из FIFO; до готовности роутера callback остаётся приостановленным.
+static void nat_peer_take(struct ll_queue* q, void* arg) {
+    struct nat_peer* peer = arg;
+    peer->pending = queue_data_get(q);
+    if (!peer->pending) { queue_resume_callback(q); return; }
+    nat_peer_send_ready(NULL, peer);
+}
 
-    new_dgram[0] = ETCP_RT_ID_NAT;
-    memcpy(new_dgram + 1, pkt->dgram + 1, ip_len);
+// Найти/создать FIFO адресата. Выделения требуются только при первом обращении к этому узлу.
+static struct nat_peer* nat_peer_get(struct UTUN_INSTANCE* inst, uint64_t node_id) {
+    struct nat_transport_ctx* tr = &inst->nat_tr;
+    struct nat_peer* peer = (struct nat_peer*)queue_find_data_by_index(tr->peers, &node_id);
+    if (peer) return peer;
+    peer = (struct nat_peer*)queue_entry_new(sizeof(*peer) - sizeof(peer->ll));
+    if (!peer) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT peer allocation failed node=%016llx", (unsigned long long)node_id); return NULL; }
+    peer->node_id = node_id; peer->inst = inst; peer->last_activity = get_time_tb();
+    peer->packets = queue_new(inst->ua, 0, 0, 0, "NAT peer packets");
+    if (!peer->packets) {
+        DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT peer queue allocation failed node=%016llx", (unsigned long long)node_id);
+        queue_entry_free(&peer->ll);
+        return NULL;
+    }
+    queue_set_callback(peer->packets, nat_peer_take, peer);
+    if (queue_data_put_with_index(tr->peers, &peer->ll) != 0) {
+        // put принимает/освобождает peer; отдельная очередь принадлежит вызывающему до успеха.
+        DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT peer index insertion failed node=%016llx", (unsigned long long)node_id);
+        return NULL;
+    }
+    return peer;
+}
 
-    struct ll_entry* new_entry = queue_entry_new(0);
-    if (!new_entry) { u_free(new_dgram); queue_dgram_free(pkt); queue_entry_free(pkt); queue_resume_callback(q); return; }
-    new_entry->dgram = new_dgram;
-    new_entry->len   = total_len;
+// 1 — пакет принят/отброшен, исходный TUN callback можно возобновить; 0 — пакет возвращён в TUN при общем лимите.
+static int nat_forward(struct UTUN_INSTANCE* inst, struct ll_entry* packet, uint64_t destination) {
+    struct nat_transport_ctx* tr = &inst->nat_tr;
+    packet->dgram[0] = ETCP_RT_ID_NAT;
+    if (tr->pending_count >= NAT_PENDING_TOTAL) {
+        tr->blocked_packet = packet; tr->blocked_destination = destination;
+        if (queue_data_put_first(tr->nat_tun->output_queue, packet) == 0) return 0;
+        tr->blocked_packet = NULL;
+        DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT failed to preserve packet at global backpressure limit");
+        return 1;
+    }
+    struct nat_peer* peer = nat_peer_get(inst, destination);
+    if (!peer) { nat_drop(inst, packet, destination, "peer allocation"); return 1; }
+    if (queue_entry_count(peer->packets) + (peer->pending != NULL) >= NAT_PENDING_PER_PEER) {
+        nat_drop(inst, packet, destination, "per-peer pending limit");
+        return 1;
+    }
+    peer->last_activity = get_time_tb();
+    tr->pending_count++;
+    if (queue_data_put(peer->packets, packet) != 0) {
+        tr->pending_count--;
+        DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT pending queue rejected packet destination=%016llx", (unsigned long long)destination);
+    }
+    return 1;
+}
 
-    queue_dgram_free(pkt); queue_entry_free(pkt);
-    queue_resume_callback(q);
+// Обе роли получают [prefix][IP]; преобразование ответа выполняется один раз до возможного ожидания.
+static void nat_tun_output(struct ll_queue* q, void* arg) {
+    struct UTUN_INSTANCE* inst = arg;
+    struct nat_transport_ctx* tr = &inst->nat_tr;
+    struct ll_entry* packet = queue_data_get(q);
+    if (!packet) { queue_resume_callback(q); return; }
+    if (!packet->dgram || packet->len < 2) {
+        nat_drop(inst, packet, 0, "invalid TUN packet");
+        queue_resume_callback(q);
+        return;
+    }
+    uint64_t destination = tr->nat_via_node_id;
+    if (packet == tr->blocked_packet) {
+        destination = tr->blocked_destination; tr->blocked_packet = NULL;
+    } else if (!destination) {
+        struct eim_nat_entry* mapping = NULL;
+        int result = eim_nat_ingress(&inst->nat, packet->dgram + 1, packet->len - 1, &mapping);
+        if (result != EIM_NAT_OK || !mapping || !mapping->src_node_id) {
+            nat_drop(inst, packet, 0, "no ingress mapping or invalid packet");
+            queue_resume_callback(q);
+            return;
+        }
+        destination = mapping->src_node_id;
+    }
+    if (nat_forward(inst, packet, destination)) queue_resume_callback(q);
+}
 
-    // После передачи роутеру new_entry и dgram больше не принадлежат NAT, даже при ошибке.
-    int send_ret = etcp_route_send(inst, TOPO_GROUP_UTUN, entry->src_node_id, new_entry, 0, 0);
-    if (send_ret != 0) {
-        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT provider: etcp_route_send back to node %016llx failed: error=%d ip_bytes=%zu",
-                   (unsigned long long)entry->src_node_id, send_ret, ip_len);
+// Пока один пакет ожидает TUN, последующие остаются в router recv_q без ACK.
+static int nat_receive_ready(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* data, size_t len) {
+    (void)group; (void)peer; (void)data; (void)len;
+    return inst->nat_tr.initialized && !inst->nat_tr.tun_pending;
+}
+
+// Повторить запись того же уже преобразованного пакета; mapping мог истечь, повторной трансляции нет.
+static void nat_tun_flush(struct UTUN_INSTANCE* inst) {
+    struct nat_transport_ctx* tr = &inst->nat_tr;
+    struct ll_entry* entry = tr->tun_pending;
+    if (!entry) return;
+    size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF;
+    if (tun_write(tr->nat_tun, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, len) < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
+        if (!tr->tun_retry_timer) tr->tun_retry_timer = uasync_set_timeout(inst->ua, NAT_WRITE_RETRY_TB, inst, nat_tun_retry, "nat_tun_retry");
+        if (!tr->tun_retry_timer) DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT TUN retry allocation failed; maintenance will retry");
+        return;
+    }
+    nat_packet_free(entry); tr->tun_pending = NULL;
+    if (inst->router_conns) {
+        for (struct ll_entry* e = inst->router_conns->head; e; e = e->next) {
+            struct ETCP_ROUTER_CONN* route = (struct ETCP_ROUTER_CONN*)e;
+            if (route->svc_id == ETCP_RT_ID_NAT && route->group_id == TOPO_GROUP_UTUN)
+                etcp_router_resume_receive(inst, route->group_id, route->remote_node_id, ETCP_RT_ID_NAT);
+        }
     }
 }
 
-// ETCP_RT_ID_NAT receive via etcp_router: CLIENT gets response, PROVIDER gets request
-static void nat_transport_etcp_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
-    if (!conn || !entry || !entry->dgram || entry->len < ROUTER_SVC_HDR_SIZE) {
-        if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
+static void nat_tun_retry(void* arg) {
+    struct UTUN_INSTANCE* inst = arg;
+    inst->nat_tr.tun_retry_timer = NULL;
+    nat_tun_flush(inst);
+}
+
+// Router передаёт владение entry; CLOSE/reset не содержит IP и не меняет таблицу.
+static void nat_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
+    if (!conn || !conn->instance || !entry || !entry->dgram || entry->len < ROUTER_SVC_HDR_SIZE) {
+        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT malformed router delivery");
+        nat_packet_free(entry);
         return;
     }
     struct UTUN_INSTANCE* inst = conn->instance;
-    if (!inst) { queue_dgram_free(entry); queue_entry_free(entry); return; }
     struct nat_transport_ctx* tr = &inst->nat_tr;
-    struct eim_nat_ctx* ctx = &inst->nat;
-    if (!tr->initialized || !ctx->initialized) { queue_dgram_free(entry); queue_entry_free(entry); return; }
-
-    uint64_t src_node_id;
-    memcpy(&src_node_id, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
-    uint64_t group_id;
-    memcpy(&group_id, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
-    if (group_id != TOPO_GROUP_UTUN || (tr->nat_via_node_id && src_node_id != tr->nat_via_node_id)) {
-        DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT receive rejected: group=%016llx source=%016llx configured_provider=%016llx",
-                   (unsigned long long)group_id, (unsigned long long)src_node_id, (unsigned long long)tr->nat_via_node_id);
-        queue_dgram_free(entry); queue_entry_free(entry);
+    if (!tr->initialized || entry->len == ROUTER_SVC_HDR_SIZE) { nat_packet_free(entry); return; }
+    uint64_t source, group;
+    memcpy(&source, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
+    memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
+    if (group != TOPO_GROUP_UTUN || (tr->nat_via_node_id && source != tr->nat_via_node_id)) {
+        nat_drop(inst, entry, source, "unexpected source/group");
         return;
     }
-    uint8_t* ip_data = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
-    size_t   ip_len  = entry->len - ROUTER_SVC_PAYLOAD_OFF;
-    if (!ip_len) { queue_dgram_free(entry); queue_entry_free(entry); return; } // Router CLOSE/reset, не IP-пакет.
-
-    if (tr->nat_via_node_id != 0) {
-        if (tr->nat_tun) {
-            if (tun_write(tr->nat_tun, ip_data, ip_len) == (ssize_t)ip_len)
-                DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT client: wrote %zu bytes from provider to TUN", ip_len);
-            else DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT client: TUN delivery failed bytes=%zu", ip_len);
-        }
-    } else {
-        int ret = eim_nat_egress(ctx, ip_data, ip_len, src_node_id, conn);
-        if (ret < 0) {
-            DEBUG_WARN(DEBUG_CATEGORY_NAT, "Egress NAT failed");
-        } else if (ret == EIM_NAT_SKIP) {
-            DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT provider: unsupported IP packet dropped bytes=%zu", ip_len);
-        } else if (ret == 0 && tr->nat_tun) {
-            if (tun_write(tr->nat_tun, ip_data, ip_len) == (ssize_t)ip_len)
-                DEBUG_DEBUG(DEBUG_CATEGORY_NAT, "NAT provider: wrote %zu bytes to internet TUN", ip_len);
-            else DEBUG_WARN(DEBUG_CATEGORY_NAT, "NAT provider: TUN delivery failed bytes=%zu", ip_len);
-        }
+    size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF;
+    if (!tr->nat_via_node_id) {
+        int result = eim_nat_egress(&inst->nat, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, len, source, conn);
+        if (result != EIM_NAT_OK) { nat_drop(inst, entry, source, "egress invalid or unsupported packet"); return; }
     }
-
-    queue_dgram_free(entry); queue_entry_free(entry);
+    if (tr->tun_pending) { nat_drop(inst, entry, source, "concurrent TUN delivery"); return; } // Возможен только loopback.
+    tr->tun_pending = entry;
+    nat_tun_flush(inst);
 }
 
-// ==================== Init / Destroy ====================
+// Освободить FIFO и waiter до уничтожения TUN-пула; вызывается только вне callbacks этого peer.
+static void nat_peer_free(struct nat_peer* peer) {
+    struct UTUN_INSTANCE* inst = peer->inst;
+    etcp_router_cancel_send_ready(inst, TOPO_GROUP_UTUN, peer->node_id, ETCP_RT_ID_NAT, &peer->waiter);
+    if (peer->pending) { nat_packet_free(peer->pending); inst->nat_tr.pending_count--; }
+    struct ll_entry* packet;
+    while ((packet = queue_data_get(peer->packets))) { nat_packet_free(packet); inst->nat_tr.pending_count--; }
+    queue_free(peer->packets);
+    queue_entry_free(&peer->ll);
+}
 
-int nat_transport_init(struct UTUN_INSTANCE* inst) {
-    if (!inst || !inst->config) return -1;
+// Периодическая работа не входит в обработку пакета. Отсутствующие после reset waiters восстанавливаются.
+static void nat_maintenance(void* arg) {
+    struct UTUN_INSTANCE* inst = arg;
     struct nat_transport_ctx* tr = &inst->nat_tr;
-    struct global_config* g = &inst->config->global;
-
-    memset(tr, 0, sizeof(*tr));
-    if (!g->nat_enabled) return 0;
-
-    tr->self_node_id    = inst->node_id;
-    tr->nat_via_node_id = g->nat_via_node_id;
-
-    if (tr->nat_via_node_id == 0) {
-        if (eim_nat_init_ctx(&inst->nat, g) != 0) {
-            DEBUG_ERROR(DEBUG_CATEGORY_NAT, "Failed to init NAT engine");
-            return -1;
+    tr->maintenance_timer = NULL;
+    uint64_t now = get_time_tb();
+    if (!tr->nat_via_node_id) eim_nat_expire(&inst->nat, now, 128);
+    for (struct ll_entry* e = tr->peers->head; e;) {
+        struct ll_entry* next = e->next;
+        struct nat_peer* peer = (struct nat_peer*)e;
+        if (peer->pending && !peer->waiter.internal && !peer->waiter.call_soon_id) nat_peer_send_ready(NULL, peer);
+        if (!peer->pending && !queue_entry_count(peer->packets) && now >= peer->last_activity && now - peer->last_activity >= 600000) {
+            queue_remove_data(tr->peers, e);
+            nat_peer_free(peer);
         }
-    } else {
-        inst->nat.initialized = 1;
+        e = next;
     }
-
-    const char* tun_name = g->nat_tun_ifname[0] ? g->nat_tun_ifname : "tun_nat";
-    char ip_str[64] = "";
-    if (g->nat_tun_ip.family == AF_INET) {
-        snprintf(ip_str, sizeof(ip_str), "%s", ip_to_str(&g->nat_tun_ip.addr.v4, AF_INET).str);
-    } else if (g->tun_ip.family == AF_INET) {
-        snprintf(ip_str, sizeof(ip_str), "%s", ip_to_str(&g->tun_ip.addr.v4, AF_INET).str);
-    } else {
-        DEBUG_ERROR(DEBUG_CATEGORY_NAT, "No NAT TUN IP");
-        eim_nat_destroy_ctx(&inst->nat);
-        return -1;
-    }
-    tr->nat_tun = tun_init_nat(inst->ua, tun_name, ip_str, TUN_MTU_DEFAULT, g->tun_test_mode);
-    if (!tr->nat_tun) {
-        DEBUG_ERROR(DEBUG_CATEGORY_NAT, "Failed to create NAT TUN %s ip=%s", tun_name, ip_str);
-        eim_nat_destroy_ctx(&inst->nat);
-        return -1;
-    }
-
-    if (etcp_router_bind(inst, ETCP_RT_ID_NAT, nat_transport_etcp_recv_cb) != 0) {
-        DEBUG_ERROR(DEBUG_CATEGORY_NAT, "Failed to bind ETCP_RT_ID_NAT via etcp_router");
-        tun_close(tr->nat_tun); tr->nat_tun = NULL;
-        eim_nat_destroy_ctx(&inst->nat);
-        return -1;
-    }
-
-    struct eim_nat_ctx* ctx = &inst->nat;
-    if (tr->nat_via_node_id != 0) {
-        if (tr->nat_tun->output_queue) queue_set_callback(tr->nat_tun->output_queue, nat_transport_client_tun_out_cb, inst);
-        DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT client via node %016llx, TUN=%s gw=%s ports=%u-%u",
-                   (unsigned long long)tr->nat_via_node_id, tun_name,
-                   ip_host_to_str(ctx->gateway_ip).str, ctx->port_start, ctx->port_end);
-    } else {
-        if (tr->nat_tun->output_queue) queue_set_callback(tr->nat_tun->output_queue, nat_transport_provider_tun_out_cb, inst);
-        DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT provider TUN=%s gw=%s ports=%u-%u",
-                   tun_name, ip_host_to_str(ctx->gateway_ip).str, ctx->port_start, ctx->port_end);
-    }
-
-    tr->initialized = 1;
-    return 0;
+    nat_tun_flush(inst);
+    tr->maintenance_timer = uasync_set_timeout(inst->ua, NAT_MAINTENANCE_TB, inst, nat_maintenance, "nat_maintenance");
+    if (!tr->maintenance_timer) DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT maintenance scheduling failed");
 }
 
+// Все ресурсы создаются в потоке ядра; при любой ошибке инициализация откатывает только свои ресурсы.
+int nat_transport_init(struct UTUN_INSTANCE* inst) {
+    if (!inst || !inst->config) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT init requires instance/config"); return -1; }
+    struct nat_transport_ctx* tr = &inst->nat_tr;
+    struct global_config* config = &inst->config->global;
+    memset(tr, 0, sizeof(*tr));
+    if (!config->nat_enabled) return 0;
+    tr->self_node_id = inst->node_id; tr->nat_via_node_id = config->nat_via_node_id;
+    if (!tr->nat_via_node_id) {
+        if (eim_nat_init_ctx(&inst->nat, config)) return -1;
+    } else inst->nat.initialized = 1;
+    const struct IP* ip = config->nat_tun_ip.family == AF_INET ? &config->nat_tun_ip : &config->tun_ip;
+    if (ip->family != AF_INET) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT requires IPv4 TUN address"); goto failed; }
+    const char* name = config->nat_tun_ifname[0] ? config->nat_tun_ifname : "tun_nat";
+    tr->nat_tun = tun_init_nat(inst->ua, name, ip_to_str(&ip->addr.v4, AF_INET).str, TUN_MTU_DEFAULT, config->tun_test_mode);
+    if (!tr->nat_tun) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT TUN creation failed name=%s", name); goto failed; }
+    tr->peers = queue_new(inst->ua, 256, 0, sizeof(uint64_t), "NAT peers");
+    if (!tr->peers) { DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT peer index allocation failed"); goto failed; }
+    if (etcp_router_bind(inst, ETCP_RT_ID_NAT, nat_receive)) goto failed;
+    tr->initialized = 1;
+    etcp_router_set_receive_ready(inst, ETCP_RT_ID_NAT, nat_receive_ready);
+    queue_set_callback(tr->nat_tun->output_queue, nat_tun_output, inst);
+    tr->maintenance_timer = uasync_set_timeout(inst->ua, NAT_MAINTENANCE_TB, inst, nat_maintenance, "nat_maintenance");
+    if (!tr->maintenance_timer) {
+        DEBUG_ERROR(DEBUG_CATEGORY_NAT, "NAT maintenance timer allocation failed");
+        nat_transport_destroy(inst);
+        return -1;
+    }
+    DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT %s initialized TUN=%s peer_limit=%u total_limit=%u",
+               tr->nat_via_node_id ? "client" : "provider", name, NAT_PENDING_PER_PEER, NAT_PENDING_TOTAL);
+    return 0;
+failed:
+    if (tr->peers) queue_free(tr->peers);
+    if (tr->nat_tun) tun_close(tr->nat_tun);
+    eim_nat_destroy_ctx(&inst->nat);
+    memset(tr, 0, sizeof(*tr));
+    return -1;
+}
+
+// Отменить производителей/таймеры и вернуть удержанные TUN-пакеты до уничтожения их пула.
 void nat_transport_destroy(struct UTUN_INSTANCE* inst) {
     if (!inst || !inst->nat_tr.initialized) return;
     struct nat_transport_ctx* tr = &inst->nat_tr;
-
+    if (tr->maintenance_timer) uasync_cancel_timeout(inst->ua, tr->maintenance_timer);
+    if (tr->tun_retry_timer) uasync_cancel_timeout(inst->ua, tr->tun_retry_timer);
     etcp_router_unbind(inst, ETCP_RT_ID_NAT);
-
-    if (tr->nat_tun) { tun_close(tr->nat_tun); tr->nat_tun = NULL; }
-
+    size_t cancelled = tr->pending_count;
+    struct ll_entry* peer;
+    while ((peer = queue_data_get(tr->peers))) nat_peer_free((struct nat_peer*)peer);
+    queue_free(tr->peers);
+    nat_packet_free(tr->tun_pending);
+    tun_close(tr->nat_tun);
     eim_nat_destroy_ctx(&inst->nat);
-
+    DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT destroyed: cancelled=%zu dropped=%llu", cancelled, (unsigned long long)tr->dropped);
     memset(tr, 0, sizeof(*tr));
-    DEBUG_INFO(DEBUG_CATEGORY_NAT, "NAT transport destroyed");
 }
diff --git a/src/nat_transport.h b/src/nat_transport.h
index 57204246..c8e571a6 100644
--- a/src/nat_transport.h
+++ b/src/nat_transport.h
@@ -16,13 +16,27 @@ extern "C" {
 
 
 #include <stdint.h>
+#include <stddef.h>
+
+#define NAT_PENDING_PER_PEER 64    // Дополнительный буфер одного получателя; переполнение локально отбрасывается с диагностикой.
+#define NAT_PENDING_TOTAL 1024     // Общий предел: остановить чтение TUN до освобождения места.
 
 struct tun_if;
+struct ll_entry;
+struct ll_queue;
 
 struct nat_transport_ctx {
     uint64_t self_node_id;    // Копия локального ID, установленная при init.
     uint64_t nat_via_node_id; // 0 — провайдер; иначе ID единственного NAT-провайдера клиента.
     struct tun_if* nat_tun;  // Отдельный принадлежащий модулю TUN, MTU=TUN_MTU_DEFAULT.
+    struct ll_queue* peers;  // Внутренний хеш получателей с раздельными FIFO и router waiters.
+    struct ll_entry* blocked_packet; // При общем backpressure остаётся в output_queue TUN.
+    uint64_t blocked_destination; // Зафиксирован до ожидания/истечения mapping.
+    size_t pending_count;    // Пакеты в FIFO получателей и их активных ожиданиях, <= NAT_PENDING_TOTAL.
+    struct ll_entry* tun_pending; // Единственный принятый IP-пакет, ожидающий доступности TUN для записи.
+    void* maintenance_timer; // Порционная очистка mapping, восстановление ожиданий и уборка получателей.
+    void* tun_retry_timer;   // Повтор записи TUN при EAGAIN; отменяется до tun_close.
+    uint64_t dropped, last_drop_log; // Счётчик отбрасываний и ограничение частоты диагностики перегрузки.
     int initialized;        // TUN, binding и callback созданы; 0 при отключённом NAT/после destroy.
 };