Add cancellable Router send requests at transport readiness

src/routing_layer/etcp_router.c

@@ -484,6 +484,7 @@ static void router_send_queue_free(void* arg) { queue_free(arg); }
 // Отменить таймеры/waiter и освободить все очереди rconn (общий код для close/restart).
 static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) {
     struct UTUN_INSTANCE* inst = rconn->inst;
+    if (rconn->send_request) etcp_router_cancel_request(rconn->send_request);
     if (rconn->receive_pump) { uasync_call_soon_cancel(inst->ua, rconn->receive_pump); rconn->receive_pump = NULL; }
     if (rconn->handshake_timer) { uasync_cancel_timeout(inst->ua, rconn->handshake_timer); rconn->handshake_timer = NULL; }
     if (rconn->ack_timer)      { uasync_cancel_timeout(inst->ua, rconn->ack_timer); rconn->ack_timer = NULL; }
@@ -594,16 +595,16 @@ static void router_set_no_route(struct ETCP_ROUTER_CONN* rconn) {
 // Попытаться продвинуть отправку: зарегистрировать waiter на send_input_q, если есть что слать.
 // Инвариант: send_q непуст ∧ !closed ∧ !no_route ∧ inflight свободен ⇒ waiter зарегистрирован.
 static void router_send_kick(struct ETCP_ROUTER_CONN* rconn) {
-    if (rconn->closed || rconn->no_route) return;
+    if (rconn->closed || rconn->no_route || rconn->send_producing) return;
     if (!rconn->peer_reset_id) {
-        if (queue_entry_count(rconn->send_q)) {
+        if (queue_entry_count(rconn->send_q) || rconn->send_request) {
             router_handshake_start(rconn);
         }
         return;
     }
     int retry = rconn->retrans_seq != rconn->retrans_end;
     if (rconn->peer_receive_blocked && !retry) return;
-    if (!retry && queue_entry_count(rconn->send_q) == 0) return;
+    if (!retry && queue_entry_count(rconn->send_q) == 0 && !rconn->send_request) return;
     if (!retry && queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) {
         rconn->send_blocked = 1;
         return;
@@ -643,6 +644,23 @@ static void router_send_drain_cb(struct ll_queue* q, void* arg) {
         rconn->send_blocked = 1;
         return;
     }
+    if (!rconn->send_q->head && rconn->send_request) {
+        struct router_send_request *h = rconn->send_request;
+        void (*ready)(struct ETCP_ROUTER_CONN*, void*) = h->ready;
+        void *context = h->arg;
+        uint64_t local = rconn->reset_id, peer = rconn->peer_reset_id;
+        DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router producer ready svc=%u waited_us=%llu inflight=%d/%u",
+                    rconn->svc_id, (unsigned long long)(get_time_us() - h->since_us),
+                    queue_entry_count(rconn->inflight_q), router_effective_max_inflight(rconn));
+        rconn->send_request = NULL; memset(h, 0, sizeof(*h));
+        rconn->send_producing = 1;
+        ready(rconn, context);
+        rconn->send_producing = 0;
+        if (rconn->closed) return;
+        if (rconn->reset_id != local || rconn->peer_reset_id != peer || router_send_conn(rconn) != conn) {
+            router_send_kick(rconn); return;
+        }
+    }
     struct ll_entry* pending = rconn->send_q->head;
     if (!pending) { rconn->send_blocked = 0; router_send_watchdog_disarm(rconn); return; }
     struct ll_entry* e = queue_entry_new(0);
@@ -697,17 +715,17 @@ static void router_send_watchdog_cb(void* arg) {
     struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg;
     rconn->watchdog_timer = NULL;
     if (rconn->closed) return;
-    if (queue_entry_count(rconn->send_q) == 0) return;
+    if (queue_entry_count(rconn->send_q) == 0 && !rconn->send_request) return;
     // Маршрут мог смениться, пока waiter остаётся на занятой очереди прежнего next hop.
     if (rconn->peer_reset_id && !rconn->no_route) router_send_kick(rconn);
-    if (queue_entry_count(rconn->send_q) > 0 && !rconn->watchdog_timer)
+    if ((queue_entry_count(rconn->send_q) > 0 || rconn->send_request) && !rconn->watchdog_timer)
         rconn->watchdog_timer = uasync_set_timeout(rconn->inst->ua,
             ROUTER_SEND_WATCHDOG_TB, rconn, router_send_watchdog_cb, "router_send_watchdog");
 }
 
 // Взвести watchdog пока send_q непуст (страховка от пропущенного re-arm).
 static void router_send_watchdog_arm(struct ETCP_ROUTER_CONN* rconn) {
-    if (rconn->watchdog_timer || queue_entry_count(rconn->send_q) == 0) return;
+    if (rconn->watchdog_timer || (queue_entry_count(rconn->send_q) == 0 && !rconn->send_request)) return;
     rconn->watchdog_timer = uasync_set_timeout(rconn->inst->ua,
         ROUTER_SEND_WATCHDOG_TB, rconn, router_send_watchdog_cb, "router_send_watchdog");
 }
@@ -910,7 +928,7 @@ static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn) {
     if (new_limit < old && inflight >= new_limit) {
         if (!rconn->send_blocked) rconn->send_blocked = 1;
     }
-    if (new_limit > old && (rconn->send_blocked || queue_entry_count(rconn->send_q) > 0)) {
+    if (new_limit > old && (rconn->send_blocked || queue_entry_count(rconn->send_q) > 0 || rconn->send_request)) {
         rconn->send_blocked = 0;
         router_send_kick(rconn);
     }
@@ -1680,6 +1698,31 @@ int etcp_router_tx_stats(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t pe
     return 0;
 }
 
+void etcp_router_cancel_request(struct router_send_request* h) {
+    if (!h || !h->conn) return;
+    struct ETCP_ROUTER_CONN *rconn = h->conn;
+    DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router producer cancelled svc=%u waited_us=%llu",
+                rconn->svc_id, (unsigned long long)(get_time_us() - h->since_us));
+    rconn->send_request = NULL; memset(h, 0, sizeof(*h));
+    if (!rconn->send_q->head && rconn->retrans_seq == rconn->retrans_end) router_cancel_send_waiter(rconn);
+    if (!rconn->send_q->head) router_send_watchdog_disarm(rconn);
+}
+
+int etcp_router_request_send(struct ETCP_ROUTER_CONN* rconn, struct router_send_request* h,
+                             void (*ready)(struct ETCP_ROUTER_CONN*, void*), void* arg) {
+    if (!rconn || rconn->closed || !h || !ready) {
+        DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router producer registration: missing/closed channel, handle or callback"); return -1;
+    }
+    if ((h->conn && h->conn != rconn) || (rconn->send_request && rconn->send_request != h)
+        || (h->conn && (h->ready != ready || h->arg != arg))) {
+        DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router producer registration conflicts svc=%u", rconn->svc_id); return -1;
+    }
+    if (h->conn) return 0;
+    *h = (struct router_send_request){rconn, ready, arg, get_time_us()}; rconn->send_request = h;
+    router_send_kick(rconn); router_send_watchdog_arm(rconn);
+    return 0;
+}
+
 // Backpressure: зарегистрировать waiter на send_q для (group_id, node_id, svc_id).
 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,

src/routing_layer/etcp_router.h

@@ -87,6 +87,16 @@ struct SVC_ROUTE_HDR {
 
 #define TRANSIT_QUEUE_HASH 256
 
+/* Одноразовый запрос слота отправки. Обнулить перед использованием; живёт до callback/отмены.
+ * Только uasync. Router снимает связь до callback; close/reset автоматически отменяет запрос. */
+struct ETCP_ROUTER_CONN;
+struct router_send_request {
+    struct ETCP_ROUTER_CONN *conn; /* Владелец регистрации, NULL после снятия. */
+    void (*ready)(struct ETCP_ROUTER_CONN*, void*); /* Формирует не более одной кодограммы через conn_send. */
+    void *arg; /* Заимствованный контекст сервиса. */
+    uint64_t since_us; /* Начало ожидания слота, без сброса при повторной регистрации. */
+};
+
 // Очередь транзитных пакетов для пары (group_id, src_node_id, dst_node_id)
 // Создаётся когда не получается отправить пакет в ETCP напрямую,
 // удаляется когда очередь становится пустой.
@@ -140,6 +150,8 @@ struct ETCP_ROUTER_CONN {
     void*    idle_ack_timer;      // idle таймер (дослать последний ack)
 
     struct ll_queue* send_q;     // очередь ожидающих отправки (inflight полон)
+    struct router_send_request *send_request; /* Ленивый producer: payload ещё принадлежит сервису. */
+    uint8_t send_producing;      /* Запрет рекурсивного kick внутри ready callback. */
     struct queue_waiter_handle send_waiter; // waiter на send_input_q (по 1 пакету, round-robin)
     struct ll_queue* send_waiter_q; // очередь регистрации, не результат повторного поиска маршрута
     uint32_t retrans_seq;         // следующий seq текущего прохода ретрансмитов
@@ -322,6 +334,17 @@ void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn);
 // В minRTT probe лимит равен 4; reset сессии возвращает default=256.
 void etcp_router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max);
 
+/* Запросить готовность handshake/маршрута/окна/ETCP при пустой send_q. 0/-1 с логом.
+ * Callback всегда отложен, разделяет round-robin с обычными очередями, может отменить/перезапустить канал.
+ * Один запрос на канал; повтор того же запроса не меняет место в очереди. В callback можно поставить
+ * одну кодограмму через conn_send и зарегистрировать следующий запрос. Вызов conn_send в этом callback
+ * только принимает копию: не вызывает сервисы синхронно. Нельзя сохранять готовность после возврата.
+ * h и arg должны жить до callback/отмены; reset/close обнуляет h->conn без вызова ready. */
+int etcp_router_request_send(struct ETCP_ROUTER_CONN* rconn, struct router_send_request* h,
+                             void (*ready)(struct ETCP_ROUTER_CONN*, void*), void* arg);
+/* Отменить исходную регистрацию; обнулённый h допустим, callback после возврата не придёт. */
+void etcp_router_cancel_request(struct router_send_request* h);
+
 // Дождаться порога send_q. h должен быть обнулён и жить до callback/отмены.
 // Callback может выполниться внутри вызова. Повторная регистрация отменяет прежнюю.
 // Если канала нет, ожидание не регистрируется. Перед освобождением arg отменить waiter.