Fix TCP connect timer ownership and pending FIN state

lib/tcp_io.c

@@ -5,11 +5,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <errno.h>
-#ifndef _WIN32
-#include <unistd.h>
-#include <sys/socket.h>
-#include <fcntl.h>
-#endif
+#include <limits.h>
 #ifndef MSG_NOSIGNAL
 #define MSG_NOSIGNAL 0
 #endif
@@ -119,8 +115,8 @@ void tcp_conn_destroy(struct tcp_conn* tc) {
     if (!tc) return;
     if (tc->destroyed) return;
     tc->destroyed = 1;
-    DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_conn_destroy: fd=%d connected=%d error=%d fin_remote=%d fin_local=%d closed=%d",
-               (int)tc->sock, tc->connected, tc->error, tc->fin_remote, tc->fin_local, tc->closed);
+    DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_conn_destroy: fd=%d connected=%d error=%d fin_remote=%d fin_pending=%d fin_local=%d closed=%d",
+               (int)tc->sock, tc->connected, tc->error, tc->fin_remote, tc->fin_pending, tc->fin_local, tc->closed);
 
     if (tc->connect_timer) { uasync_cancel_timeout(tc->ua, tc->connect_timer); tc->connect_timer = NULL; }
 
@@ -153,6 +149,7 @@ void tcp_conn_destroy(struct tcp_conn* tc) {
 static void tcp_conn_handle_error(struct tcp_conn* tc, int err)
 {
     tc->error = 1;
+    if (tc->connect_timer) { uasync_cancel_timeout(tc->ua, tc->connect_timer); tc->connect_timer = NULL; }
 
     queue_waiter_cancel(tc->read_queue, &tc->read_waiter);
     queue_set_empty_callback(tc->read_queue, NULL, NULL);
@@ -325,7 +322,14 @@ static void write_queue_fetch_cb(struct ll_queue* q, void* arg) {
         } else if (e->dgram == &tcp_fin_sentinel) {
             DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: FIN fd=%d", (int)tc->sock);
             queue_entry_free(e); queue_resume_callback(q);
-            shutdown(tc->sock, SHUT_WR);
+            tc->fin_pending = 0;
+            if (shutdown(tc->sock, SHUT_WR) != 0) {
+                int err = socket_get_error();
+                DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: shutdown write failed fd=%d err=%d (%s)",
+                            (int)tc->sock, err, socket_strerror(err));
+                tcp_conn_handle_error(tc, err);
+                return;
+            }
             tc->fin_local = 1;
             if (tc->on_fin_sent) tc->on_fin_sent(tc, tc->arg);
         } else {
@@ -467,11 +471,13 @@ static void error_cb(socket_t sock, void* arg) {
 
 int tcp_conn_push_fin(struct tcp_conn* tc) {
     if (!tc || tc->sock == SOCKET_INVALID) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin invalid tc"); return -1; }
+    if (tc->fin_pending) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin already pending fd=%d", (int)tc->sock); return -1; }
     if (tc->fin_local) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin already fin_local fd=%d", (int)tc->sock); return -1; }
     if (tc->closed) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin already closed fd=%d", (int)tc->sock); return -1; }
     struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool);
     if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin entry_pool exhausted fd=%d", (int)tc->sock); return -1; }
     e->dgram = (uint8_t*)&tcp_fin_sentinel; e->len = 0;
+    tc->fin_pending = 1;
     DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: push_fin fd=%d wq=%d", (int)tc->sock, tc->write_queue->count + 1);
     queue_data_put(tc->write_queue, e);
     return 0;
@@ -514,11 +520,42 @@ static void connect_timeout_cb(void* arg) {
     }
 }
 
-void tcp_conn_set_connect_timeout(struct tcp_conn* tc, int timeout_ms) {
-    if (!tc) return;
-    if (timeout_ms < 0) timeout_ms = 0;
-    tc->connect_timeout_ms = timeout_ms;
-    if (!tc->connected && timeout_ms > 0) {
-        tc->connect_timer = uasync_set_timeout(tc->ua, timeout_ms * 10, tc, connect_timeout_cb, "tcp_conn_ct");
+/* Замена атомарна относительно event loop: сначала резервируем новый таймер, затем отменяем прежний. */
+int tcp_conn_set_connect_timeout(struct tcp_conn* tc, int timeout_ms) {
+    if (!tc) {
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: set connect timeout: tc=NULL timeout=%d", timeout_ms);
+        return -EINVAL;
     }
+    if (tc->destroyed || tc->error || tc->closed || tc->sock == SOCKET_INVALID) {
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: set connect timeout on inactive tc fd=%d destroyed=%d error=%d closed=%d",
+                    (int)tc->sock, tc->destroyed, tc->error, tc->closed);
+        return -EINVAL;
+    }
+    if (timeout_ms > INT_MAX / 10) {
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: connect timeout=%d exceeds limit=%d ms fd=%d",
+                    timeout_ms, INT_MAX / 10, (int)tc->sock);
+        return -EINVAL;
+    }
+    if (timeout_ms < 0) timeout_ms = 0;
+    void* timer = NULL;
+    if (!tc->connected && timeout_ms > 0) {
+        timer = uasync_set_timeout(tc->ua, timeout_ms * 10, tc, connect_timeout_cb, "tcp_conn_ct");
+        if (!timer) {
+            DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot create connect timer fd=%d timeout=%d ms; old timeout=%d retained",
+                        (int)tc->sock, timeout_ms, tc->connect_timeout_ms);
+            return -ENOMEM;
+        }
+    }
+    void* old_timer = tc->connect_timer;
+    if (old_timer && uasync_cancel_timeout(tc->ua, old_timer) != ERR_OK) {
+        DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot cancel connect timer fd=%d timer=%p", (int)tc->sock, old_timer);
+        if (timer && uasync_cancel_timeout(tc->ua, timer) != ERR_OK)
+            DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: cannot cancel replacement connect timer fd=%d timer=%p", (int)tc->sock, timer);
+        return -EIO;
+    }
+    tc->connect_timer = timer;
+    tc->connect_timeout_ms = timeout_ms;
+    DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "tcp_io: connect timeout updated fd=%d connected=%d timeout=%d ms old_timer=%p timer=%p",
+                (int)tc->sock, tc->connected, timeout_ms, old_timer, timer);
+    return 0;
 }

lib/tcp_io.h

@@ -9,7 +9,8 @@
 //   FIN (recv==0) → on_fin (сразу или через empty_callback после дренажа очереди).
 //
 // Запись: внешний код кладёт entry в write_queue через queue_data_put().
-//   write_queue_fetch_cb (deferred) → send(), при EAGAIN → write_buf + EPOLLOUT.
+//   Первый write_queue_fetch_cb может быть синхронным; продолжение — через uasync_call_soon.
+//   send(), при EAGAIN → write_buf + EPOLLOUT.
 //   Всё отправлено → on_flushed.
 //
 // FIN/Close: tcp_conn_push_fin/close ставят сентинел в write_queue.
@@ -49,6 +50,7 @@ struct tcp_conn {
     uint8_t  connected;
     uint8_t  error;
     uint8_t  fin_remote;              // FIN получен от удалённой стороны (recv == 0)
+    uint8_t  fin_pending;             // FIN поставлен в write_queue, shutdown ещё не выполнен
     uint8_t  fin_local;               // FIN отправлен удалённой стороне (shutdown SHUT_WR)
     uint8_t  closed;                  // сокет полностью закрыт (close)
     uint8_t  destroyed;               // 1 = tcp_conn_destroy вызван, tc ожидает отложенного free
@@ -91,37 +93,44 @@ struct tcp_conn* tcp_conn_create(
     void (*on_error)(struct tcp_conn* tc, int err, void* arg),
     void* arg);
 
-// Дренирует очереди, закрывает сокет, уничтожает пулы, освобождает tc.
+// Сразу отменяет callbacks/таймер, дренирует очереди и закрывает сокет.
+// Пулы и tc освобождаются через uasync_call_soon в том же uasync-потоке.
+// NULL безопасен; повторный destroy допустим только до отложенного free.
+// Владелец после destroy больше не использует tc; event loop должен выполнить отложенную очистку.
 void tcp_conn_destroy(struct tcp_conn* tc);
 
 // Запись: queue_data_put(tc->write_queue, entry) напрямую.
-// Автозабор (deferred) сам шлёт когда сокет готов.
+// Первый автозабор может быть синхронным; queue_resume_callback откладывает продолжение.
 // Порог: 32 записи, при заполнении queue_waiter_wait.
 
 // Поставить сигналы в очередь отправки. Все данные в очереди перед сигналом
-// будут отправлены до его обработки. Кодирование: dgram=NULL → close, len=0+dgram!=NULL → FIN.
+// будут переданы ОС до его обработки. Сигналы имеют len=0: dgram=NULL → CLOSE,
+// dgram=&tcp_fin_sentinel → FIN (внутренний указатель, сигналы ставить только через API).
 //   tcp_conn_push_fin(tc)    — FIN: после отправки предшествующих данных вызывает
 //                              shutdown(SHUT_WR) и on_fin_sent.
-//                              Повторный вызов игнорируется (fin_sent уже установлен).
+//                              Повторный вызов возвращает -1 при fin_pending или fin_local.
 //   tcp_conn_push_close(tc)  — CLOSE: после отправки предшествующих данных закрывает
 //                              сокет (close), вызывает on_closed.
 //                              После close сокет удалён из uasync, tc->closed=1.
-//                              Повторный вызов игнорируется.
+//                              После фактического закрытия повторный вызов возвращает -1.
+// Оба API: 0 = сигнал принят; -1 = ошибка. Callback может выполниться до возврата из push.
 int tcp_conn_push_fin(struct tcp_conn* tc);
 int tcp_conn_push_close(struct tcp_conn* tc);
 
 // Одноразовый коллбэк: вызывается когда write_queue + write_buf полностью опустели.
-// После вызова сбрасывается. Установить повторно можно в любой момент.
+// Сбрасывается перед вызовом; callback может установить новое уведомление.
 void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_conn* tc, void* arg));
 
 // Принудительная остановка чтения: убирает EPOLLIN и отменяет read_waiter.
 void tcp_conn_pause_read(struct tcp_conn* tc);
 
-// Таймаут установления соединения (мс, 0 = без таймаута). Если соединение ещё не
-// установлено (connected=0), взводит таймер; по истечении вызывает
-// tcp_conn_handle_error(ETIMEDOUT) → on_error. Таймер гасится при успешном connect
-// и при tcp_conn_destroy.
-void tcp_conn_set_connect_timeout(struct tcp_conn* tc, int timeout_ms);
+// Заменить connect-таймаут: мс от текущего вызова, <=0 отменяет таймер.
+// Только для живого tc в его uasync-потоке. Для connected=1 таймер не создаётся.
+// Допустимый положительный диапазон: <= INT_MAX/10 (uasync принимает 0.1 мс).
+// Возврат: 0 = применено; -EINVAL = аргументы/состояние, -ENOMEM = создание таймера,
+// -EIO = нарушение состояния отменяемого таймера. При ошибке прежняя настройка сохраняется.
+// Функция не вызывает callbacks. Истечение → on_error(ETIMEDOUT); отмена при connect/error/destroy.
+int tcp_conn_set_connect_timeout(struct tcp_conn* tc, int timeout_ms);
 
 
 #ifdef __cplusplus