From 232ad2de6b65d366b916722c9ad9896155eb8c3a Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 5 Sep 2026 22:34:00 +0300 Subject: [PATCH] =?UTF-8?q?etcp=5Frouter:=20=D1=80=D0=B0=D0=B7=D0=B4=D0=B5?= =?UTF-8?q?=D0=BB=D0=B8=D1=82=D1=8C=20=D1=81=D0=BB=D1=83=D0=B6=D0=B5=D0=B1?= =?UTF-8?q?=D0=BD=D1=8B=D0=B5=20=D1=84=D1=83=D0=BD=D0=BA=D1=86=D0=B8=D0=B8?= =?UTF-8?q?=20=D0=B8=20API=20=D0=B2=20=D1=85=D0=B5=D0=B4=D0=B5=D1=80=D0=B5?= =?UTF-8?q?=20+=20=D1=87=D0=B8=D1=81=D1=82=D0=BA=D0=B0=20=D0=BC=D1=91?= =?UTF-8?q?=D1=80=D1=82=D0=B2=D0=BE=D0=B3=D0=BE=20=D0=BA=D0=BE=D0=B4=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - etcp_router.h: 4 секции (протокол / внутренние структуры / API / служебные) - переименование router_set_max_inflight -> etcp_router_set_max_inflight - удалены неиспользуемые: input_q_count, waiter_register/cancel, conn_close_async (+ поля close_callback/close_callback_arg) - docs: актуализировать backpressure (on_send_ready на send_q) вместо удалённого waiter API --- src/proxy/PROTOCOL.md | 45 ++++++++++++------------ src/proxy/socks_proxy_doc.md | 2 +- src/proxy/tcp_proxy_client_doc.md | 6 ++-- src/routing_layer/etcp_router.c | 40 ++-------------------- src/routing_layer/etcp_router.h | 51 ++++++++++++++-------------- src/routing_layer/etcp_router_doc.md | 8 +---- 6 files changed, 55 insertions(+), 97 deletions(-) diff --git a/src/proxy/PROTOCOL.md b/src/proxy/PROTOCOL.md index 4e189ac0..d1958458 100644 --- a/src/proxy/PROTOCOL.md +++ b/src/proxy/PROTOCOL.md @@ -66,9 +66,9 @@ subcmd: ДОП. СОСТОЯНИЯ: -┌─────────────────┐ send_data в ETCP вернул -1 (normalizer полон) +┌─────────────────┐ send_data в ETCP вернул -1 (send_q полон) │ TX_BACKPRESSURE│ не tcp_recved() → окно lwIP=0 → recv_cb остановлен -│ tx_buf != NULL │ etcp_router_waiter_register() → ждём освобождения +│ tx_buf != NULL │ etcp_router_on_send_ready() → ждём освобождения │ tx_waiter pend │ waiter_cb: retry send_data → tcp_recved → окно открыто └─────────────────┘ @@ -147,7 +147,7 @@ subcmd: ДОП. СОСТОЯНИЯ: ┌─────────────────┐ send_msg в ETCP вернул -1 │ PAUSED │ буферизуем в pause_buf, убираем read_id (пауза чтения) -│ pause_buf>0 │ etcp_router_waiter_register() → ждём освобождения +│ pause_buf>0 │ etcp_router_on_send_ready() → ждём освобождения │ pause_waiter │ waiter_cb: retry send_msg → resume read_id └─────────────────┘ @@ -186,7 +186,7 @@ subcmd: │ │ recv_cb │ │──CONNECT+DATA──► │ │handle_connect│ │ │ │(lwIP→ETCP) │ │ │ │handle_data │ │ │ │ │ │ backpressure: │ └──────┬───────┘ │ - │ │ send fail: │ │ waiter на normalizer │ │ sock_send │ + │ │ send fail: │ │ waiter на send_q │ │ sock_send │ │ │ tx_buf+ │ │ │ ┌──────▼───────┐ │ │ │ waiter │ │ │ │ out_buf │ │ │ │ окно lwIP=0│ │ │ │ pending_buf │ │ @@ -214,10 +214,10 @@ subcmd: ``` ══════════════════════════════════════════════════════════════════════ - Используется встроенный механизм ll_queue threshold waiter: - etcp_router_waiter_register(inst, peer_node_id, &waiter) - → внутри: queue_waiter_wait(normalizer->input_queue, &waiter) - etcp_router_waiter_cancel(inst, peer_node_id, &waiter) + Используется встроенный механизм ll_queue threshold waiter (на router send_q): + etcp_router_on_send_ready(inst, group_id, node_id, svc_id, &waiter, cb, arg) + → внутри: queue_waiter_wait(rconn->send_q, &waiter) + etcp_router_cancel_send_ready(inst, group_id, node_id, svc_id, &waiter) CLIENT (lwIP → ETCP): recv_cb(buf, len): @@ -226,7 +226,7 @@ subcmd: if ret < 0: tx_buf = copy(buf), tx_len = len НЕ tcp_recved() → окно lwIP = 0, recv_cb больше не вызывается - etcp_router_waiter_register(&tx_waiter) + etcp_router_on_send_ready(inst, group_id, via_node_id, ETCP_RT_ID_TCP_PROXY_SERVER, &tx_waiter, tx_waiter_cb, pc) tx_waiter callback: ret = send_data(tx_buf, tx_len) @@ -234,7 +234,7 @@ subcmd: tcp_recved(tx_len) → окно открыто, lwIP возобновляет recv_cb free(tx_buf) if tun_closed → send_close() - // если ret < 0 → waiter остаётся, normalizer вызовет снова + // если ret < 0 → waiter остаётся, router вызовет снова SERVER (socket → ETCP): read_cb(buf, n): @@ -243,7 +243,7 @@ subcmd: if ret < 0: pause_buf = copy(buf), pause_len = n uasync_remove_socket_t(sock) → пауза чтения с сокета - etcp_router_waiter_register(&pause_waiter) + etcp_router_on_send_ready(inst, group_id, peer_node_id, ETCP_RT_ID_TCP_PROXY_CLIENT, &pause_waiter, pause_waiter_cb, rc) pause_waiter callback: ret = send_msg(DATA, pause_buf, pause_len) @@ -433,7 +433,7 @@ tcp_proxy_client_conn: pcb — lwIP tcp_pcb tx_buf — буфер при backpressure lwIP→ETCP (send_data fail) tx_len — длина tx_buf - tx_waiter — ll_queue waiter на normalizer.input_queue + tx_waiter — waiter на router send_q (etcp_router_on_send_ready) tcp_proxy_server_conn: stream_id — уникальный ID потока @@ -449,7 +449,7 @@ tcp_proxy_server_conn: out_backoff — backoff для retry send в сокет pause_buf — буфер при backpressure socket→ETCP (send_msg fail) pause_len — длина pause_buf - pause_waiter — ll_queue waiter на normalizer.input_queue + pause_waiter — waiter на router send_q (etcp_router_on_send_ready) close_timer — таймер повтора CLOSE/ERROR close_backoff — backoff для retry CLOSE/ERROR sock — OS сокет к адресату @@ -462,16 +462,17 @@ tcp_proxy_server_conn: ``` // etcp_router.h -void etcp_router_waiter_register(struct UTUN_INSTANCE* inst, - uint64_t peer_node_id, struct queue_waiter_handle* h); -void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, - uint64_t peer_node_id, struct queue_waiter_handle* h); +void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, + uint64_t node_id, uint8_t svc_id, struct queue_waiter_handle* h, + queue_threshold_callback_fn cb, void* arg); +void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, + uint64_t node_id, uint8_t svc_id, struct queue_waiter_handle* h); // Внутри: -// conn = route_bgp_find_conn_for_node(peer_node_id) -// queue_waiter_wait(conn->normalizer->input, h) -// queue_waiter_cancel(conn->normalizer->input, h) +// rconn = router_conn_find(inst, group_id, node_id, svc_id) +// queue_waiter_wait(rconn->send_q, h, cb, arg) +// queue_waiter_cancel(rconn->send_q, h) // -// normalizer.input_queue уже имеет threshold=0 (ждёт полного освобождения) -// через queue_set_threshold(input_queue, 0, 0) в etcp.c +// send_q — очередь router conn при полном inflight +// (threshold = ROUTER_MAX_SEND_Q_PACKETS - 1 в router_conn_reset) ``` diff --git a/src/proxy/socks_proxy_doc.md b/src/proxy/socks_proxy_doc.md index 55d58ad1..8e2b534b 100644 --- a/src/proxy/socks_proxy_doc.md +++ b/src/proxy/socks_proxy_doc.md @@ -124,7 +124,7 @@ URL переписывается в относительный (`/path`), ост ## 5. Зависимости - `tcp_proxy_server.h` — константы subcmd (`TCP_PROXY_SUBCMD_*`) и размер заголовка (`TCP_PROXY_HDR_SIZE=6`) - `etcp.h` / `etcp_api.h` — `etcp_route_send()`, `ETCP_RT_ID_TCP_PROXY` -- `etcp_router.h` — `etcp_router_on_send_ready()`, `etcp_router_cancel_send_ready()`, `etcp_router_waiter_cancel()` +- `etcp_router.h` — `etcp_router_on_send_ready()`, `etcp_router_cancel_send_ready()` - `tcp_io.h` / `tcp_conn` — TCP-соединения (`tcp_conn_create`, `tcp_conn_push_close`, `tcp_conn_push_fin`, write/read очереди) - `u_async.h` — event loop, `uasync_add_socket_t` - `ll_queue.h` — очереди (write_queue, read_queue), waiter механизм diff --git a/src/proxy/tcp_proxy_client_doc.md b/src/proxy/tcp_proxy_client_doc.md index dcba7551..caef21de 100644 --- a/src/proxy/tcp_proxy_client_doc.md +++ b/src/proxy/tcp_proxy_client_doc.md @@ -47,7 +47,7 @@ tcp_proxy_client_create(inst, ua, tun_name, tun_ip, mtu, test_mode, **Client-side lwIP** — чтение данных (`tcp_proxy_client_recv_cb`): - При получении данных от lwIP (клиент что-то отправляет): копируем pbuf → `tcp_proxy_client_send_data()` → ETCP DATA -- При backpressure (очередь нормализатора переполнена): сохраняем в `tx_buf`, регистрируем waiter через `etcp_router_waiter_register()`. При освобождении очереди `tcp_proxy_client_tx_waiter_cb` досылает данные и вызывает `tcp_recved()` для window update +- При backpressure (router send_q переполнен): сохраняем в `tx_buf`, регистрируем waiter через `etcp_router_on_send_ready()`. При освобождении очереди `tcp_proxy_client_tx_waiter_cb` досылает данные и вызывает `tcp_recved()` для window update - При p==NULL (FIN от lwIP): выставляется `fin_local`, отправляется ETCP FIN. Если обе стороны закрылись — ETCP CLOSE **Exit-side data** — приём ответных данных (`tcp_proxy_client_handle_data()`): @@ -90,7 +90,7 @@ struct tcp_proxy_client_mapping_config { ### Механизмы надёжности -- **Backpressure в нормализатор**: `etcp_router_waiter_register()` — если `etcp_route_send()` не может отправить (очередь переполнена), данные буферизуются в `tx_buf` и досылаются при освобождении +- **Backpressure на router send_q**: `etcp_router_on_send_ready()` — если `etcp_route_send()` не может отправить (inflight полон), данные буферизуются в `tx_buf` и досылаются при освобождении - **Повтор CLOSE/ERROR**: если отправка CLOSE/ERROR не удалась, выставляется `close_pending=1`. В `poll_cb` повторяется попытка - **Очередь to_lwip**: буферизует данные от exit до готовности lwIP send buffer, обеспечивая window-based flow control @@ -180,7 +180,7 @@ struct tcp_proxy_client_mapping_config { - **`tcp_proxy_client_tun_input()`** — коллбэк очереди output_queue TUN. Парсит IP-пакеты: TCP → lwip_tcp_input (с динамическим listen на целевой порт), UDP/ICMP → соответствующие прокси - **`tcp_proxy_client_handle_non_tcp()`** — перенаправление UDP в `udp_proxy_send_to_exit()` и ICMP Echo в `icmp_proxy_send_to_exit()` - **`tcp_proxy_client_feed_from_transport()`** — чтение данных из очереди `to_lwip` и запись в lwIP через `tcp_write()` пока есть место в send buffer. После feed вызывает `tcp_output()` -- **`tcp_proxy_client_tx_waiter_cb()`** — коллбэк backpressure waiter: досылка `tx_buf` при освобождении normalizer очереди, вызов `tcp_recved()` для window update +- **`tcp_proxy_client_tx_waiter_cb()`** — коллбэк backpressure waiter: досылка `tx_buf` при освобождении router send_q, вызов `tcp_recved()` для window update - **`tcp_proxy_client_ensure_outbound_listen()`** — динамическое создание listen pcb в lwIP для заданного порта (если ещё не создан), нужно для прозрачного проксирования любых исходящих TCP-соединений #### Управление памятью и соединениями diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index aa12dc8d..1eb6124d 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -307,10 +307,7 @@ static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) { // Закрыть conn: отправить CLOSE удалённой стороне, уведомить локальный сервис, очистить static void router_close_finalize(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; - void (*cb)(void*) = rconn->close_callback; - void* cb_arg = rconn->close_callback_arg; queue_entry_free(&rconn->ll); - if (cb) cb(cb_arg); } // Отменить таймеры/waiter и освободить все очереди rconn (общий код для close/restart). @@ -350,7 +347,7 @@ static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) { } } -// Пересоздать очереди и сбросить всё состояние rconn (кроме identity: group/remote/svc/ll.data/inst/sess_id/close_cb/last_dgram_ts). +// Пересоздать очереди и сбросить всё состояние rconn (кроме identity: group/remote/svc/ll.data/inst/sess_id/last_dgram_ts). static void router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { router_conn_free_queues(rconn); @@ -653,7 +650,7 @@ static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn) { } } -void router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max) { +void etcp_router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max) { if (!rconn) return; if (new_max < MINRTT_PROBE_MAX_INFLIGHT) new_max = MINRTT_PROBE_MAX_INFLIGHT; rconn->max_inflight = new_max; @@ -1222,30 +1219,6 @@ int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_ return ret; } -int etcp_router_input_q_count(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id) { - if (!inst || !topo_groups_find(inst->topo_groups, group_id)) return -1; - struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, group_id), node_id); - if (!conn || !conn->normalizer || !conn->normalizer->input) return -1; - return queue_entry_count(conn->normalizer->input); -} - -void etcp_router_waiter_register(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t peer_node_id, - struct queue_waiter_handle* h, - queue_threshold_callback_fn callback, void* arg) { - if (!inst || !topo_groups_find(inst->topo_groups, group_id) || !h) return; - struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, group_id), peer_node_id); - if (!conn || !conn->normalizer || !conn->normalizer->input) return; - queue_waiter_wait(conn->normalizer->input, h, callback, arg); -} - -void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t peer_node_id, - struct queue_waiter_handle* h) { - if (!inst || !topo_groups_find(inst->topo_groups, group_id) || !h) return; - struct ETCP_CONN* conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, group_id), peer_node_id); - if (!conn || !conn->normalizer || !conn->normalizer->input) return; - queue_waiter_cancel(conn->normalizer->input, h); -} - void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id, uint8_t svc_id) { struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, group_id, remote_node_id, svc_id); if (!rconn) return; @@ -1347,15 +1320,6 @@ void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn) { router_close_and_notify(rconn); } -void etcp_router_conn_close_async(struct ETCP_ROUTER_CONN* rconn, - void (*on_close_done)(void* arg), - void* close_arg) { - if (!rconn) return; - rconn->close_callback = on_close_done; - rconn->close_callback_arg = close_arg; - router_close_and_notify(rconn); -} - // Reset retransmit state for all router connections to a node // (no callbacks, no close notifications — lightweight, for conn reinit) void etcp_router_pause_retrans_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id) { diff --git a/src/routing_layer/etcp_router.h b/src/routing_layer/etcp_router.h index 9c8776d4..ad231366 100644 --- a/src/routing_layer/etcp_router.h +++ b/src/routing_layer/etcp_router.h @@ -18,7 +18,9 @@ extern "C" { #include #include "etcp_api.h" -#define TRANSIT_QUEUE_HASH 256 +// ═══════════════════════════════════════════════════════════════ +// 1. Протокол SVC_ROUTE (wire-формат и формат доставки в сервис) +// ═══════════════════════════════════════════════════════════════ #pragma pack(push, 1) struct SVC_ROUTE_HDR { @@ -52,6 +54,12 @@ struct SVC_ROUTE_HDR { #define ROUTER_FLAG_SIGNED 0x08 // пакет содержит Ed25519 подпись (64 байта после payload) #define ROUTER_FLAG_CLOSE 0x02 // нормальное закрытие conn +// ═══════════════════════════════════════════════════════════════ +// 2. Внутренние структуры и константы (etcp_router.c, control_server.c) +// ═══════════════════════════════════════════════════════════════ + +#define TRANSIT_QUEUE_HASH 256 + // Очередь транзитных пакетов для пары (group_id, src_node_id, dst_node_id) // Создаётся когда не получается отправить пакет в ETCP напрямую, // удаляется когда очередь становится пустой. @@ -110,8 +118,6 @@ struct ETCP_ROUTER_CONN { void* no_route_timer; // таймер 20ms проверки появления маршрута uint8_t no_ack_count; // счётчик последовательных ретрансмиссий без ACK uint8_t closed; // 1 = в процессе закрытия, таймеры игнорируют - void* close_callback; // пользовательский коллбэк завершения закрытия - void* close_callback_arg; // аргумент для close_callback uint32_t sess_id; // локальный счётчик переустановок (0 при создании, ++ при рестарте; на wire не пишется) uint8_t start_sent; // 0 = нужно отправить START в первом data uint8_t peer_sync_done; // 0 = rx_seq ещё не синхронизирован с удалённым seq (авто-создание/рестарт) @@ -173,6 +179,10 @@ struct ETCP_ROUTER_BINDINGS { etcp_recv_fn callbacks[SVC_ROUTE_MAX_BINDINGS]; }; +// ═══════════════════════════════════════════════════════════════ +// 3. Публичное API (для сервисов) +// ═══════════════════════════════════════════════════════════════ + // Инициализация: etcp_bind(ETCP_ID_SVC_ROUTE) + очистка bindings + создание router_conns int etcp_router_init(struct UTUN_INSTANCE* inst); @@ -200,29 +210,7 @@ int etcp_router_conn_send(struct ETCP_ROUTER_CONN* rconn, void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn); // Установить рабочий max_inflight, пересчитывает inflight_limit и state machine -void router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max); - -// Асинхронное закрытие: коллбэк вызывается после освобождения памяти rconn -void etcp_router_conn_close_async(struct ETCP_ROUTER_CONN* rconn, - void (*on_close_done)(void* arg), - void* close_arg); - -// Сбросить ретрансмиты для всех router_conn к узлу (без колбэков, без CLOSE) -// Используется при etcp_conn_reinit чтобы избежать гонки с очисткой ETCP-очередей -void etcp_router_pause_retrans_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id); - -// Закрыть все router_conn для указанного (group_id, remote_node_id) (peer умер) -void etcp_router_conn_close_all_for_node(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id); - -// Возвращает количество пакетов в очереди normalizer->input для узла node_id в группе group_id -int etcp_router_input_q_count(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id); - -// Backpressure: зарегистрировать/отменить waiter на normalizer->input очереди -void etcp_router_waiter_register(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t peer_node_id, - struct queue_waiter_handle* h, - queue_threshold_callback_fn callback, void* arg); -void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t peer_node_id, - struct queue_waiter_handle* h); +void etcp_router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max); // Backpressure: зарегистрировать/отменить waiter на send_q очереди void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id, @@ -234,6 +222,17 @@ void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id // Backpressure: проверить есть ли место в send_q (без регистрации waiter) int etcp_router_send_q_has_room(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id); +// ═══════════════════════════════════════════════════════════════ +// 4. Служебные функции (для каркаса: etcp.c, topo_group.c) +// ═══════════════════════════════════════════════════════════════ + +// Сбросить ретрансмиты для всех router_conn к узлу (без колбэков, без CLOSE) +// Используется при etcp_conn_reinit чтобы избежать гонки с очисткой ETCP-очередей +void etcp_router_pause_retrans_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id); + +// Закрыть все router_conn для указанного (group_id, remote_node_id) (peer умер) +void etcp_router_conn_close_all_for_node(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id); + // Сбросить состояние роутера для конкретного peer+svc в группе (перезапуск удалённой стороны). void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id, uint8_t svc_id); diff --git a/src/routing_layer/etcp_router_doc.md b/src/routing_layer/etcp_router_doc.md index c2bfb5a0..2ba6d644 100644 --- a/src/routing_layer/etcp_router_doc.md +++ b/src/routing_layer/etcp_router_doc.md @@ -48,9 +48,6 @@ void my_service_callback(struct ETCP_CONN* conn, struct ll_entry* entry) { struct queue_waiter_handle h; etcp_router_on_send_ready(inst, group_id, node_id, svc_id, &h, my_ready_cb, my_arg); // Когда send_q освободится — вызовется my_ready_cb - -// На normalizer->input (сетевая очередь): -etcp_router_waiter_register(inst, group_id, node_id, &h, my_ready_cb, my_arg); ``` ### Подпись / шифрование (модуль route_crypto) @@ -130,13 +127,10 @@ ACK: периодический (10ms) + idle (500ms) → router_send_ack(rx_seq | `etcp_router_conn_get(inst, group, remote, svc_id)` | Найти или создать ROUTER_CONN | | `etcp_router_conn_send(rconn, data, len, mode)` | Отправить данные с авто-seq и inflight-контролем | | `etcp_router_conn_close(rconn)` | Закрыть seq-подключение (CLOSE + уведомление сервиса) | -| `etcp_router_conn_close_async(rconn, cb, arg)` | Асинхронное закрытие с коллбэком | | `etcp_router_conn_close_all_for_node(inst, group, node_id)` | Закрыть все conn к узлу | | `etcp_router_conn_restart(inst, group, remote, svc_id)` | Сброс состояния (перезапуск удалённой стороны) | | `etcp_router_pause_retrans_for_node(inst, node_id)` | Сбросить ретрансмиты без закрытия (для conn reinit) | -| `router_set_max_inflight(rconn, new_max)` | Установить рабочий max_inflight | -| `etcp_router_input_q_count(inst, group, node_id)` | Размер normalizer->input очереди узла | -| `etcp_router_waiter_register/cancel(inst, group, node_id, h, cb, arg)` | Backpressure на normalizer->input | +| `etcp_router_set_max_inflight(rconn, new_max)` | Установить рабочий max_inflight | | `etcp_router_on_send_ready/cancel_send_ready(inst, group, node, svc, h, cb, arg)` | Backpressure на send_q | | `etcp_router_transit_queues_destroy(conn)` | Удалить все транзитные очереди ETCP_CONN |