Own STCP adapter queues and report connection failures before READY

src/transport_layer/stcp_link.c

@@ -42,10 +42,8 @@ struct stcp_link {
     void (*on_close_cb)(struct stcp_link *link, int err, void *arg);
     void *close_arg;
 
-    struct ll_queue *tx_queue;     // owned by this link
-
-    struct ll_queue *saved_rx_queue;  // saved rx_queue for pre-closed conn cleanup
-    uint8_t conn_pre_closed;          // 1 = conn already CLOSED before stcp_link_close
+    struct ll_queue *rx_queue;     // Обе очереди принадлежат адаптеру, conn только заимствует их.
+    struct ll_queue *tx_queue;
 
     uint64_t peer_reset_id;            // received from peer during handshake (reset epoch)
     uint8_t peer_device_type;          // CLIENT_TYPE_* from peer handshake
@@ -56,6 +54,7 @@ struct stcp_link {
 
 static void link_rx_cb(struct ll_queue *q, void *arg) {
     struct stcp_link *link = (struct stcp_link *)arg;
+    if (link->closing) return;
     struct ll_entry *e = queue_data_get(q);
     if (!e) { queue_resume_callback(q); return; }
 
@@ -88,19 +87,45 @@ static void link_rx_cb(struct ll_queue *q, void *arg) {
             inst->api_bindings.callbacks[ETCP_ID_DATA](link->etcp_conn ? link->etcp_conn : NULL, e);
         else { queue_dgram_free(e); queue_entry_free(e); }
     }
-    queue_resume_callback(q);
-    if (link->conn) stcp_rx_resume_if_needed(link->conn);
+    if (!link->closing) {
+        queue_resume_callback(q);
+        if (link->conn) stcp_rx_resume_if_needed(link->conn);
+    }
 }
 
 // ====== STCP close → link down ======
 
 static void stcp_link_on_stcp_close(struct stcp_conn *conn, int err, void *arg) {
     struct stcp_link *link = (struct stcp_link *)arg;
-    (void)conn; (void)err;
     callring_put(CR_SLINK_ONSTCP, (uintptr_t)conn, (uintptr_t)arg);
     link->ready = 0;
-    if (link->etcp_link) { link->etcp_link->recv_keepalive = 0; link->etcp_link->link_status = 0; }
-    if (link->on_close_cb) link->on_close_cb(link, err, link->close_arg);
+    link->conn = NULL; /* Серверный conn освобождается после возврата из этого callback. */
+    conn->rx_queue = conn->tx_queue = NULL;
+    queue_set_callback(link->rx_queue, NULL, NULL);
+    queue_set_callback(link->tx_queue, NULL, NULL);
+    void (*callback)(struct stcp_link*, int, void*) = link->on_close_cb;
+    void *callback_arg = link->close_arg;
+    link->on_close_cb = NULL; link->close_arg = NULL;
+    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "stcp adapter closed: link=%p error=%d cancelling=%u", (void*)link, err, link->closing);
+    if (!link->closing && callback) callback(link, err, callback_arg);
+    /* Callback мог вызвать close; после уведомления адаптер больше не используем. */
+}
+
+static int stcp_link_init_queues(struct stcp_link *link) {
+    link->rx_queue = queue_new(link->ua, 0, 0, 0, "stcp_rx");
+    link->tx_queue = queue_new(link->ua, 0, 0, 0, "stcp_tx");
+    if (!link->rx_queue || !link->tx_queue) {
+        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP adapter queue allocation failed link=%p RX=%p TX=%p",
+            (void*)link, (void*)link->rx_queue, (void*)link->tx_queue);
+        return -1;
+    }
+    queue_set_callback(link->rx_queue, link_rx_cb, link);
+    queue_set_threshold(link->rx_queue, STCP_RX_QUEUE_MAX_PACKETS, STCP_RX_QUEUE_MAX_BYTES);
+    queue_set_threshold(link->tx_queue, 0, 0);
+    queue_set_waiter_defer(link->tx_queue, 1);
+    stcp_conn_set_rx_queue(link->conn, link->rx_queue);
+    stcp_conn_set_tx_queue(link->conn, link->tx_queue);
+    return 0;
 }
 
 // ====== server accept → link ======
@@ -108,7 +133,7 @@ static void stcp_link_on_stcp_close(struct stcp_conn *conn, int err, void *arg)
 static void server_accept_cb(struct stcp_conn *conn, void *arg) {
     struct stcp_server *ss = (struct stcp_server *)arg;
     struct stcp_link *link = u_calloc(1, sizeof(struct stcp_link));
-    if (!link) { stcp_conn_free(conn); return; }
+    if (!link) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP server adapter allocation failed"); stcp_conn_do_close(conn, ENOMEM); return; }
     link->inst = ss->inst;
     link->ua = ss->inst->ua;
     link->ready = 1;
@@ -119,14 +144,7 @@ static void server_accept_cb(struct stcp_conn *conn, void *arg) {
 
     stcp_conn_set_on_close(conn, stcp_link_on_stcp_close, link);
 
-    struct ll_queue *rx = queue_new(conn->ua, 0, 0, 0, "srx");
-    queue_set_callback(rx, link_rx_cb, link);
-    queue_set_threshold(rx, STCP_RX_QUEUE_MAX_PACKETS, STCP_RX_QUEUE_MAX_BYTES);
-    stcp_conn_set_rx_queue(conn, rx);
-    link->tx_queue = queue_new(conn->ua, 0, 0, 0, "stx");
-    queue_set_threshold(link->tx_queue, 0, 0);
-    queue_set_waiter_defer(link->tx_queue, 1);
-    stcp_conn_set_tx_queue(conn, link->tx_queue);
+    if (stcp_link_init_queues(link) < 0) { stcp_link_close(link); return; }
 
     DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_link: server accepted connection");
     if (ss->on_link) ss->on_link(link, ss->tcp_sock);
@@ -136,23 +154,14 @@ static void server_accept_cb(struct stcp_conn *conn, void *arg) {
 
 static void client_ready_cb(struct stcp_conn *conn, void *arg) {
     struct stcp_link *link = (struct stcp_link *)arg;
-    if (!conn) return;
-    link->ready = 1;
+    if (!conn || link->closing) return;
     link->conn = conn;
     link->peer_reset_id = conn->peer_reset_id;
     link->peer_device_type = conn->peer_device_type;
     link->peer_keepalive_interval = conn->peer_keepalive_interval;
 
-    stcp_conn_set_on_close(conn, stcp_link_on_stcp_close, link);
-
-    struct ll_queue *rx = queue_new(conn->ua, 0, 0, 0, "crx");
-    queue_set_callback(rx, link_rx_cb, link);
-    queue_set_threshold(rx, STCP_RX_QUEUE_MAX_PACKETS, STCP_RX_QUEUE_MAX_BYTES);
-    stcp_conn_set_rx_queue(conn, rx);
-    link->tx_queue = queue_new(conn->ua, 0, 0, 0, "ctx");
-    queue_set_threshold(link->tx_queue, 0, 0);
-    queue_set_waiter_defer(link->tx_queue, 1);
-    stcp_conn_set_tx_queue(conn, link->tx_queue);
+    if (stcp_link_init_queues(link) < 0) { stcp_conn_do_close(conn, ENOMEM); return; }
+    link->ready = 1;
 
     {
         const struct sockaddr_storage *ra = link->etcp_link ? &link->etcp_link->remote_addr : NULL;
@@ -163,6 +172,7 @@ static void client_ready_cb(struct stcp_conn *conn, void *arg) {
                    (unsigned long long)link->peer_reset_id);
     }
     if (link->etcp_link) etcp_link_enter_ready_tcp(link->etcp_link);
+    if (link->closing || !link->etcp_conn || link->etcp_conn->close_requested) return;
     if (link->on_ready_cb) link->on_ready_cb(link, link->ready_arg);
 }
 
@@ -272,6 +282,13 @@ struct stcp_link *stcp_link_connect(struct ETCP_LINK *etcp_link,
                                      0,
                                      client_ready_cb, link, NULL, NULL, NULL, NULL, local_addr, 0, reality, socks, inst);
     if (!link->cli) { u_free(link); return NULL; }
+    link->conn = stcp_client_get_conn(link->cli);
+    if (!link->conn || link->conn->state == STCP_STATE_CLOSED || link->conn->state == STCP_STATE_ERROR) {
+        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP adapter startup failed before return target=%s:%u", addr_str, rport);
+        stcp_client_destroy(link->cli); u_free(link); return NULL;
+    }
+    /* Установить до первого poll: закрытие connect/handshake также должно достигать владельца. */
+    stcp_conn_set_on_close(link->conn, stcp_link_on_stcp_close, link);
 
     DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_link: connecting to %s:%u pubkey=%016llx bind=%s",
                addr_str, rport, (unsigned long long)*(const uint64_t*)pubkey,
@@ -296,32 +313,15 @@ static void stcp_link_close_impl(void *arg) {
     struct stcp_link *link = (struct stcp_link *)arg;
     callring_put(CR_SLINK_IMPL, (uintptr_t)link, (uintptr_t)(link->conn));
 
-    if (link->cli) {
-        struct stcp_conn *c = stcp_client_get_conn(link->cli);
-        if (!link->conn_pre_closed) {
-            if (c && c->rx_queue) { stcp_drain_free_queue(c->rx_queue); c->rx_queue = NULL; }
-            if (link->tx_queue) { stcp_drain_free_queue(link->tx_queue); link->tx_queue = NULL; }
-            stcp_client_destroy(link->cli);
-        } else {
-            if (link->saved_rx_queue) { stcp_drain_free_queue(link->saved_rx_queue); link->saved_rx_queue = NULL; }
-            if (link->tx_queue) { stcp_drain_free_queue(link->tx_queue); link->tx_queue = NULL; }
-            u_free(link->cli);
-        }
-    } else if (link->conn) {
-        if (!link->conn_pre_closed) {
-            struct stcp_conn *conn = link->conn;
-            if (conn->rx_queue) { stcp_drain_free_queue(conn->rx_queue); conn->rx_queue = NULL; }
-            if (link->tx_queue) { stcp_drain_free_queue(link->tx_queue); link->tx_queue = NULL; }
-            stcp_conn_do_close(conn, 0);
-        } else {
-            /* conn уже закрыт и освобождается отложенно — не трогаем link->conn,
-             * работаем с сохранённой rx_queue. */
-            if (link->saved_rx_queue) { stcp_drain_free_queue(link->saved_rx_queue); link->saved_rx_queue = NULL; }
-            if (link->tx_queue) { stcp_drain_free_queue(link->tx_queue); link->tx_queue = NULL; }
-        }
-        link->conn = NULL;
+    struct stcp_conn *conn = link->conn;
+    if (conn) {
+        conn->rx_queue = conn->tx_queue = NULL;
+        conn->on_close = NULL; conn->close_arg = NULL;
     }
-    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "stcp_link_close_impl: link=%p cli=%p conn=%p conn_pre_closed=%d", (void*)link, (void*)link->cli, (void*)link->conn, link->conn_pre_closed);
+    stcp_drain_free_queue(link->rx_queue); stcp_drain_free_queue(link->tx_queue);
+    if (link->cli) stcp_client_destroy(link->cli);
+    else if (conn) stcp_conn_do_close(conn, 0);
+    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "stcp adapter freed: link=%p client=%u", (void*)link, link->cli != NULL);
     u_free(link);
 }
 
@@ -329,6 +329,10 @@ void stcp_link_close(struct stcp_link *link) {
     if (!link) return;
     if (link->closing) return;
     link->closing = 1;
+    link->ready = 0;
+    link->on_ready_cb = NULL; link->ready_arg = NULL;
+    queue_set_callback(link->rx_queue, NULL, NULL);
+    queue_set_callback(link->tx_queue, NULL, NULL);
     callring_put(CR_SLINK_CLOSE, (uintptr_t)link, (uintptr_t)(link->etcp_link));
     /* Обрыв вперёд-ссылки родителя: после отложенного free (close_impl)
      * etcp_link->tcp_link не должен остаться висячим. */
@@ -342,44 +346,44 @@ void stcp_link_close(struct stcp_link *link) {
     link->close_arg = NULL;
     DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "stcp_link_close: link=%p closing=%d", (void*)link, link->closing);
 
-    if (link->cli) {
-        struct stcp_conn *c = stcp_client_get_conn(link->cli);
-        if (c) {
-            if (c->hs_timer) { uasync_cancel_timeout(link->ua, c->hs_timer); c->hs_timer = NULL; }
-            if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) {
-                link->saved_rx_queue = c->rx_queue; c->rx_queue = NULL; link->conn_pre_closed = 1;
-            }
-        }
-    } else if (link->conn) {
+    if (link->conn) {
         struct stcp_conn *c = link->conn;
         if (c->hs_timer) { uasync_cancel_timeout(link->ua, c->hs_timer); c->hs_timer = NULL; }
-        if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) {
-            link->saved_rx_queue = c->rx_queue; c->rx_queue = NULL; link->conn_pre_closed = 1;
+        if (c->socket_id) {
+            uasync_set_socket_read(link->ua, c->socket_id, 0);
+            uasync_set_socket_write(link->ua, c->socket_id, 0);
         }
     }
-    uasync_call_soon(link->ua, link, stcp_link_close_impl);
+    if (!uasync_call_soon(link->ua, link, stcp_link_close_impl)) {
+        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP adapter mandatory teardown scheduling failed link=%p", (void*)link);
+        abort();
+    }
 }
 
 int stcp_link_send(struct stcp_link *link, const uint8_t *data, size_t len) {
-    if (!link || !link->ready) return -1;
+    if (!link || link->closing || !link->ready || !link->conn) return -1;
     struct ll_entry *e = queue_entry_new(0);
     if (!e) return -1;
     e->dgram = u_malloc(len ? len : 1);
     if (!e->dgram) { queue_entry_free(e); return -1; }
     if (len) memcpy(e->dgram, data, len);
     e->len = (uint16_t)len;
-    queue_data_put(link->conn->tx_queue, e);
+    queue_data_put(link->tx_queue, e);
     return 0;
 }
 
 int stcp_link_is_ready(struct stcp_link *link) {
-    return link ? link->ready : 0;
+    return link && !link->closing ? link->ready : 0;
 }
 
 struct ETCP_CONN *stcp_link_get_etcp_conn(struct stcp_link *link) {
     return link ? link->etcp_conn : NULL;
 }
 
+struct ETCP_LINK *stcp_link_get_etcp_link(struct stcp_link *link) {
+    return link && !link->closing ? link->etcp_link : NULL;
+}
+
 void stcp_link_set_etcp_conn(struct stcp_link *link, struct ETCP_CONN *conn) {
     if (!link) return;
     link->etcp_conn = conn;

src/transport_layer/stcp_link.h

@@ -1,7 +1,9 @@
 /* stcp_link — адаптер STCP-соединения к ETCP_LINK: handshake, очереди DATA, ready/close callbacks.
  * Серверный путь создаёт входящие линки, connect запускает исходящий для заданного ETCP_LINK.
  * send копирует data в tx_queue (0 / -1); ready означает готовность STCP, не READY topo-группы.
- * close гасит callback внезапного разрыва и откладывает teardown; после close handle не использовать.
+ * Адаптер владеет RX/TX очередями; stcp_conn их заимствует до закрытия.
+ * close сразу запрещает send/RX/READY, гасит callback внезапного разрыва и откладывает teardown;
+ * повторный close внутри активного callback допустим, после возврата в poll handle не использовать.
  * API и callbacks — uasync-поток. Указатели get_* заимствованы; адреса peer/local используют static-буферы. */
 #ifndef STCP_LINK_H
 #define STCP_LINK_H
@@ -52,11 +54,15 @@ struct stcp_server *stcp_server_list_find_by_sock(struct UTUN_INSTANCE *inst, co
 
 struct stcp_link *stcp_link_connect(struct ETCP_LINK *etcp_link,
                                     struct sockaddr_storage *remote_addr, uint16_t port);
+/* NULL — запуск не состоялся, callback не вызывается. Принятый connect уведомляет также об ошибке до READY.
+ * Все уведомления после возврата connect; владелец устанавливает callbacks до следующего poll. */
 void stcp_link_close(struct stcp_link *link);
 int  stcp_link_send(struct stcp_link *link, const uint8_t *data, size_t len);
 int  stcp_link_is_ready(struct stcp_link *link);
 
 struct ETCP_CONN *stcp_link_get_etcp_conn(struct stcp_link *link);
+/* Заимствованный физический владелец; NULL после close. Проверять после callback до доступа к ETCP_LINK. */
+struct ETCP_LINK *stcp_link_get_etcp_link(struct stcp_link *link);
 void stcp_link_set_etcp_conn(struct stcp_link *link, struct ETCP_CONN *conn);
 void stcp_link_set_etcp_link(struct stcp_link *link, struct ETCP_LINK *elink);
 const struct sockaddr_storage *stcp_link_get_remote_addr(struct stcp_link *link);