feat: send_q backpressure + ll_queue on_get callback + consumer_ack flow control

etcp_router: ROUTER_MAX_SEND_Q_PACKETS=256, router_enqueue_send returns
error when send_q full, etcp_route_send propagates -1 on backpressure.

ll_queue: queue_set_on_get — deferred callback after each queue_data_get.
Replaces manual consumer_ack polling with queue-driven notification.

tcp_proxy_server: consumer_ack=1, write_on_get_cb triggers ACK on
each write_queue dequeue. Full backpressure chain:
  write_queue full → no dequeue → no ACK → sender stops → all queues pause
  write_queue drains → dequeue → on_get → ACK → sender resumes

lib/ll_queue.c

@@ -47,6 +47,7 @@ static inline void queue_check_thread(struct ll_queue* q) {
 static void queue_resume_timeout_cb(void* arg);
 static void check_waiters(struct ll_queue* q);
 static void empty_trampoline(void* arg);
+static void on_get_trampoline(void* arg);
 
 static void empty_trampoline(void* arg) {
     struct ll_queue* q = (struct ll_queue*)arg;
@@ -58,6 +59,11 @@ static void empty_trampoline(void* arg) {
     if (q->count == 0 && cb) cb(q, cb_arg);
 }
 
+static void on_get_trampoline(void* arg) {
+    struct ll_queue* q = (struct ll_queue*)arg;
+    if (q->on_get) q->on_get(q, q->on_get_arg);
+}
+
 // ==================== Управление очередью ====================
 
 struct ll_queue* queue_new(struct UASYNC* ua, size_t hash_size, uint16_t index_offset, uint16_t index_size, char* name) {
@@ -151,6 +157,8 @@ void queue_free(struct ll_queue* q) {
     if (old_eid) uasync_call_soon_cancel(q->ua, old_eid);
     q->empty_callback = NULL;
     q->empty_callback_arg = NULL;
+    q->on_get = NULL;
+    q->on_get_arg = NULL;
     
     u_free(q);
 }
@@ -552,6 +560,9 @@ struct ll_entry* queue_data_get(struct ll_queue* q) {
     if (q->count == 0 && q->empty_callback && !q->empty_call_soon_id)
         q->empty_call_soon_id = uasync_call_soon(q->ua, q, (timeout_callback_t)empty_trampoline);
     
+    if (q->on_get)
+        uasync_call_soon(q->ua, q, (timeout_callback_t)on_get_trampoline);
+    
 #ifdef QUEUE_DEBUG
     queue_check_consistency(q);// !!!! for debug
 #endif
@@ -644,6 +655,12 @@ void queue_set_empty_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void
         q->empty_call_soon_id = uasync_call_soon(q->ua, q, (timeout_callback_t)empty_trampoline);
 }
 
+void queue_set_on_get(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg) {
+    if (!q) return;
+    q->on_get = cbk_fn;
+    q->on_get_arg = arg;
+}
+
 int queue_waiter_wait(struct ll_queue* q, struct queue_waiter_handle* h,
                       queue_threshold_callback_fn callback, void* arg) {
     if (!q || !h) return -1;

lib/ll_queue.h

@@ -148,6 +148,9 @@ struct ll_queue {
     void*             empty_callback_arg;
     void*             empty_call_soon_id; // handle от uasync_call_soon
 
+    queue_callback_fn on_get;             // коллбэк после каждого queue_data_get (deferred)
+    void*             on_get_arg;
+
     struct ll_entry** hash_table;   // Хеш-таблица для поиска по id (если hash_size > 0)
     size_t hash_size;               // Размер хеш-таблицы
     uint16_t index_offset;          // Смещение индекса в data[] (для всех entry очереди)
@@ -243,6 +246,14 @@ void queue_set_waiter_defer(struct ll_queue* q, int enable);
  */
 void queue_set_empty_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg);
 
+/**
+ * @brief Устанавливает коллбэк вызываемый после каждого queue_data_get (deferred через call_soon).
+ * @param q очередь
+ * @param cbk_fn коллбэк (NULL для отмены)
+ * @param arg пользовательский аргумент
+ */
+void queue_set_on_get(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg);
+
 /**
  * @brief Регистрирует ожидание освобождения очереди до общего порога.
  * @param q очередь

src/etcp_router.c

@@ -171,11 +171,16 @@ static void router_send_resume_cb(void* arg) {
     router_drain_send_q(rconn);
 }
 
-static void router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len) {
+static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len) {
+    if (queue_entry_count(rconn->send_q) >= ROUTER_MAX_SEND_Q_PACKETS) {
+        DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_send_q: FULL svc_id=%u count=%d — backpressure",
+                   rconn->svc_id, queue_entry_count(rconn->send_q));
+        return -1;
+    }
     struct ll_entry* qe = queue_entry_new(0);
-    if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_enqueue_send: queue_entry_new failed"); return; }
+    if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_enqueue_send: queue_entry_new failed"); return -1; }
     qe->dgram = u_malloc(pl_len);
-    if (!qe->dgram) { queue_entry_free(qe); return; }
+    if (!qe->dgram) { queue_entry_free(qe); return -1; }
     qe->len = pl_len;
     if (pl_len > 0) memcpy(qe->dgram, payload, pl_len);
     DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router_send_q: queued svc_id=%u inflight=%d send_q=%d",
@@ -185,6 +190,7 @@ static void router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* p
     if (!rconn->send_resume_timer)
         rconn->send_resume_timer = uasync_set_timeout(rconn->inst->ua,
             ROUTER_SEND_RESUME_TB, rconn, router_send_resume_cb, "router_send_resume");
+    return 0;
 }
 
 // ====================================================================
@@ -513,9 +519,9 @@ int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_
     if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return -1; }
 
     if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) {
-        router_enqueue_send(rconn, entry->dgram + 1, payload_len);
+        int eq_ret = router_enqueue_send(rconn, entry->dgram + 1, payload_len);
         queue_dgram_free(entry); queue_entry_free(entry);
-        return 0;
+        return eq_ret;
     }
 
     int ret = router_send_one(rconn, entry->dgram + 1, payload_len);
@@ -585,6 +591,7 @@ struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst,
     if (!rconn->recv_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(recv_q) failed"); queue_entry_free(&rconn->ll); return NULL; }
     rconn->send_q = queue_new(inst->ua, 0, 0, 0, "router_send_q");
     if (!rconn->send_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(send_q) failed"); queue_free(rconn->recv_q); queue_entry_free(&rconn->ll); return NULL; }
+    queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS / 2, 0);
     DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: new conn remote=%016llx svc_id=%u",
                (unsigned long long)remote_node_id, svc_id);
     queue_data_put_with_index(inst->router_conns, &rconn->ll);
@@ -599,8 +606,7 @@ int etcp_router_conn_send(struct ETCP_ROUTER_CONN* rconn,
         return -1;
     }
     if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) {
-        router_enqueue_send(rconn, data, len);
-        return 0;
+        return router_enqueue_send(rconn, data, len);
     }
     return router_send_one(rconn, data, len);
 }

src/etcp_router.h

@@ -53,6 +53,7 @@ struct ETCP_ROUTER_CONN {
 #define ROUTER_ACK_INTERVAL_TB    100    // интервал ACK: 10ms в timebase (0.1ms)
 #define ROUTER_ACK_IDLE_TB        5000   // idle таймаут: 500ms
 #define ROUTER_SEND_RESUME_TB     500    // retry интервал send_q: 50ms
+#define ROUTER_MAX_SEND_Q_PACKETS 256    // порог backpressure на send_q
 
 // Флаги в старших битах seq (взаимоисключающие)
 #define ROUTER_SEQ_CLOSE_FLAG  0x80000000  // нормальное закрытие conn

src/proxy/tcp_proxy_server.c

@@ -100,9 +100,16 @@ static void send_error(struct tcp_proxy_server_conn* rc) {
 }
 
 // ====================================================================
-// Коллбэки tcp_io
+// Коллбэки tcp_io + очередь
 // ====================================================================
 
+static void write_on_get_cb(struct ll_queue* q, void* arg) {
+    (void)q;
+    struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
+    struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
+    if (inst) etcp_router_consumer_ack(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY);
+}
+
 static void on_fin_cb(struct tcp_conn* tc, void* arg) {
     struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
     if (write_pending(tc))
@@ -112,6 +119,7 @@ static void on_fin_cb(struct tcp_conn* tc, void* arg) {
 }
 
 static void on_flushed_cb(struct tcp_conn* tc, void* arg) {
+    (void)tc;
     struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
     if (rc->cli_closed) { tcp_proxy_server_conn_free(rc); return; }
     send_close(rc);
@@ -168,7 +176,11 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) {
     if (ret == 0) {
         queue_resume_callback(q);
     } else {
-        etcp_router_waiter_register(inst, rc->peer_node_id, &rc->pause_waiter, pause_resume_cb, rc);
+        struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY);
+        if (rconn && rconn->send_q)
+            queue_waiter_wait(rconn->send_q, &rc->pause_waiter, pause_resume_cb, rc);
+        else
+            etcp_router_waiter_register(inst, rc->peer_node_id, &rc->pause_waiter, pause_resume_cb, rc);
     }
 }
 
@@ -202,7 +214,12 @@ void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc) {
     if (rc->tc) { tcp_conn_destroy(rc->tc); rc->tc = NULL; }
     if (rc->close_timer) { uasync_cancel_timeout(rc->ua, rc->close_timer); rc->close_timer = NULL; }
     if (rc->diag_timer) { uasync_cancel_timeout(rc->ua, rc->diag_timer); rc->diag_timer = NULL; }
-    if (rc->ctx && rc->ctx->inst) etcp_router_waiter_cancel(rc->ctx->inst, rc->peer_node_id, &rc->pause_waiter);
+    if (rc->ctx && rc->ctx->inst) {
+        struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(rc->ctx->inst, rc->peer_node_id, ETCP_ID_TCP_PROXY);
+        if (rconn && rconn->send_q) queue_waiter_cancel(rconn->send_q, &rc->pause_waiter);
+        else etcp_router_waiter_cancel(rc->ctx->inst, rc->peer_node_id, &rc->pause_waiter);
+    }
+    if (rc->tc) queue_set_on_get(rc->tc->write_queue, NULL, NULL);
     u_free(rc);
 }
 
@@ -254,6 +271,13 @@ int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry*
 
     rc->next = ctx->conns; ctx->conns = rc;
     rc->diag_timer = uasync_set_timeout(rc->ua, 10000, rc, diag_timer_cb, "tps_diag");
+
+    {
+        struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, src_node_id, ETCP_ID_TCP_PROXY);
+        if (rconn) rconn->consumer_ack = 1;
+    }
+    queue_set_on_get(rc->tc->write_queue, write_on_get_cb, rc);
+
     queue_dgram_free(entry); queue_entry_free(entry);
     return 0;
 }