diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index 08570803..2cff8e40 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -215,7 +215,6 @@ static int etcp_down_timer_start(struct ETCP_CONN* conn) { conn->down_timer = uasync_set_timeout(conn->instance->ua, ETCP_DOWN_TIMEOUT_TB, conn, etcp_down_timeout_cb, "etcp_down"); if (!conn->down_timer) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "[%s] cannot schedule DOWN timeout", conn->log_name); - etcp_connection_close(conn); return -1; } DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "[%s] DOWN timer started timeout_ms=%d refs=%d", @@ -317,7 +316,7 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n return NULL; } - if (etcp_down_timer_start(etcp) < 0) return NULL; + if (etcp_down_timer_start(etcp) < 0) { etcp_connection_close(etcp); return NULL; } DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] connection initialized. ETCP=%p mtu=%d, next_tx_id=%u state=pending", etcp->log_name, etcp, etcp->mtu, etcp->next_tx_id); @@ -1113,7 +1112,7 @@ void etcp_on_link_down(struct ETCP_CONN* etcp, struct ETCP_LINK* down_link) { } int was_up = etcp->links_up; etcp->links_up = up; - if (!up) etcp_down_timer_start(etcp); + if (!up && etcp_down_timer_start(etcp) < 0) { etcp_connection_close(etcp); return; } else etcp_down_timer_cancel(etcp); if (up == 0 && was_up != 0) { DEBUG_INFO(DEBUG_CATEGORY_ETCP, "All links fall down"); diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c index 76614f04..cd3a35b7 100644 --- a/src/transport_layer/etcp_connections.c +++ b/src/transport_layer/etcp_connections.c @@ -2906,7 +2906,7 @@ int init_connections(struct UTUN_INSTANCE* instance) { } } - node_conn_direct_init(instance); + if (node_conn_direct_init(instance) < 0) return -1; etcp_keepalive_register(instance); // Return 1 if there was a partial socket initialization error diff --git a/tests/test_etcp_server_lifecycle.c b/tests/test_etcp_server_lifecycle.c new file mode 100644 index 00000000..b9ec7c74 --- /dev/null +++ b/tests/test_etcp_server_lifecycle.c @@ -0,0 +1,192 @@ +/* Входящий ETCP: реальные очереди, таймеры, NCD и управляемые часы без сетевого I/O. */ +#include +#include +#include "etcp.h" +#include "etcp_api.h" +#include "etcp_keepalive.h" +#include "node_conn_direct.h" +#include "utun_instance.h" +#include "config_parser.h" +#include "test_clock.h" +#include "../lib/debug_config.h" +#include "../lib/mem.h" + +#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "FAIL %s:%d: %s", __func__, __LINE__, #x); abort(); } } while (0) + +static struct UTUN_INSTANCE inst; +static struct utun_config cfg; +static struct ETCP_SOCKET sock; +static unsigned deleted, closed, down; +static int release_on_down; + +static void deleted_cb(struct ETCP_CONN* conn, int event, void* arg) { + (void)conn; (void)event; (void)arg; deleted++; +} + +/* Освобождение последнего владельца внутри DOWN не сокращает срок сохранения ETCP. */ +static void owner_cb(struct NODE_CONN_DIRECT* handle, enum ncd_event event, void* arg) { + (void)arg; + if (event == NCD_EVENT_DOWN) { + down++; + if (release_on_down) node_conn_direct_close(handle); + } else if (event == NCD_EVENT_CLOSED) { + closed++; CHECK(node_conn_direct_get_conn(handle) == NULL); + } +} + +static void setup(void) { + memset(&inst, 0, sizeof(inst)); memset(&cfg, 0, sizeof(cfg)); memset(&sock, 0, sizeof(sock)); + deleted = closed = down = 0; release_on_down = 0; + inst.ua = uasync_create(); CHECK(inst.ua); inst.config = &cfg; + cfg.global.bbr_max_cwnd = 100000; cfg.global.keepalive_interval = 200; cfg.global.keepalive_timeout = 2000; + inst.connections = queue_new(inst.ua, 16, 0, 8, "server lifecycle"); CHECK(inst.connections); + inst.pkt_pool = memory_pool_init(sizeof(struct ETCP_DGRAM), "server lifecycle packets"); CHECK(inst.pkt_pool); + sock.instance = &inst; strcpy(sock.name, "synthetic server"); + CHECK(init_connections(&inst) == 0); // чисто входящий узел тоже принимает NCD CLOSE + CHECK(inst.ncd_control_bound && !inst.ncd_registry); +} + +static void teardown(void) { + while (inst.connections->head) { + struct conn_queue_entry* entry = (void*)inst.connections->head->data; + etcp_connection_close(entry->conn); + } + uasync_drain_immediate(inst.ua); CHECK(!inst.ncd_registry); + while (inst.activity_cbks) utun_remove_activity_cbk(&inst, inst.activity_cbks->fn, inst.activity_cbks->arg); + queue_free(inst.connections); memory_pool_destroy(inst.pkt_pool); uasync_poll(inst.ua, 0); + CHECK(inst.ua->timer_alloc_count == inst.ua->timer_free_count); + uasync_destroy(inst.ua, 0); +} + +static struct ETCP_CONN* new_conn(void) { + struct ETCP_CONN* conn = etcp_connection_create(&inst, "incoming"); CHECK(conn); + conn->peer_node_id = 42; etcp_conn_add_cbk(conn, deleted_cb, NULL, ETCP_CBK_EVENT_DELETE); + queue_set_callback(conn->send_input_q, NULL, NULL); + return conn; +} + +/* Фикстура владеет только сокетом; линки выделены штатно освобождаемой памятью ETCP. */ +static struct ETCP_LINK* add_link(struct ETCP_CONN* conn, int tcp) { + struct ETCP_LINK* link = u_calloc(1, sizeof(*link)); CHECK(link); + link->etcp = conn; link->conn = &sock; link->is_server = 1; link->is_tcp = tcp; + link->initialized = link->link_status = link->recv_keepalive = link->remote_keepalive = 1; + link->link_state = LINK_STATE_CONNECTED; link->keepalive_interval = link->ka_period_ms = 200; + link->keepalive_timeout = 2000; link->last_recv_local_time = get_time_tb(); + link->local_link_id = conn->links ? conn->links->local_link_id + 1 : 1; + link->next = conn->links; conn->links = link; + return link; +} + +static struct ETCP_CONN* ready_conn(int tcp) { + struct ETCP_CONN* conn = new_conn(); add_link(conn, tcp); etcp_conn_ready(conn); + CHECK(conn->links_up && !conn->down_timer); + return conn; +} + +static void receive_control(struct ETCP_CONN* conn, uint8_t subcmd) { + struct ncd_control_msg msg = { ETCP_RT_ID_NCD_CONTROL, subcmd, conn->peer_node_id }; + struct ll_entry* entry = ll_alloc_lldgram(sizeof(msg)); CHECK(entry); + memcpy(entry->dgram, &msg, sizeof(msg)); entry->len = sizeof(msg); + inst.api_bindings.callbacks[ETCP_RT_ID_NCD_CONTROL](conn, entry); + uasync_drain_immediate(inst.ua); +} + +/* Pending имеет конечный срок; внешняя ссылка задерживает только освобождение памяти. */ +static void check_pending(void) { + setup(); struct ETCP_CONN* conn = new_conn(); CHECK(etcp_conn_ref_take(conn) == 0); + test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB - 10000); CHECK(!deleted && inst.connections->count == 1); + test_clock_run_for(inst.ua, 20000); CHECK(deleted == 1 && conn->state == 2 && !conn->down_timer); + CHECK(!inst.connections->count && !conn->free_scheduled); + etcp_conn_ref_free(conn); teardown(); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS initial DOWN timeout and external reference"); +} + +/* Два пути, потеря последнего, неизменный deadline, восстановление и новая потеря. */ +static void check_reconnect(void) { + setup(); struct ETCP_CONN* conn = ready_conn(0); struct ETCP_LINK* second = add_link(conn, 1); + etcp_link_close(second); CHECK(conn->links_up && !conn->down_timer); + conn->links->link_status = 0; etcp_on_link_down(conn, conn->links); CHECK(conn->down_timer); + void* timer = conn->down_timer; + test_clock_run_for(inst.ua, 200000); etcp_on_link_down(conn, conn->links); CHECK(conn->down_timer == timer); + conn->links->link_status = 1; conn->link_ready_for_send_fn(conn); CHECK(conn->links_up && !conn->down_timer); + test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); CHECK(!deleted && inst.connections->count == 1); + etcp_link_close(conn->links); CHECK(!conn->links && !conn->links_up && conn->down_timer); + test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); CHECK(deleted == 1 && !inst.connections->count); + teardown(); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS multiple links, reconnect cancellation and fresh DOWN deadline"); +} + +/* Локальный handle не препятствует удалению; release внутри DOWN сохраняет таймаут. */ +static void check_owner(int release) { + setup(); struct ETCP_CONN* conn = ready_conn(0); struct NODE_CONN_DIRECT* handle = NULL; + CHECK(node_conn_direct_open(&inst, 42, owner_cb, NULL, &handle, NULL) == NCD_REUSED); + uasync_poll(inst.ua, 0); release_on_down = release; + conn->links->link_status = 0; etcp_on_link_down(conn, conn->links); + CHECK(down == 1 && !deleted && conn->down_timer && !conn->close_requested); + test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); + CHECK(deleted == 1 && closed == (release ? 0 : 1) && !inst.connections->count); + if (!release) node_conn_direct_close(handle); + teardown(); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS DOWN timeout with local owner release_on_down=%d", release); +} + +/* RX продлевает watchdog, затем молчание даёт DOWN и ещё 30 секунд до DELETE. */ +static void check_silent_peer(int tcp) { + setup(); struct ETCP_CONN* conn = ready_conn(tcp); struct ETCP_LINK* link = conn->links; + etcp_keepalive_watch_rx(link); CHECK(link->rx_timeout_timer); + test_clock_run_for(inst.ua, 15000); link->last_recv_local_time = get_time_tb(); etcp_keepalive_watch_rx(link); + test_clock_run_for(inst.ua, 10000); CHECK(conn->links_up && !conn->down_timer); + test_clock_run_for(inst.ua, 15000); CHECK(!conn->links_up && conn->down_timer && !deleted); + test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); CHECK(deleted == 1 && !inst.connections->count); + teardown(); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS silent incoming peer tcp=%d", tcp); +} + +/* Сон не даёт преждевременный DOWN, но локальные фазовые таймеры не продлевают жизнь пира. */ +static void check_sleep(void) { + setup(); struct ETCP_CONN* conn = ready_conn(1); struct ETCP_LINK* link = conn->links; + struct ETCP_DGRAM* pkt = memory_pool_alloc(inst.pkt_pool); CHECK(pkt); + memset(pkt, 0, sizeof(*pkt)); pkt->data_len = 8; pkt->data[0] = ETCP_KEEPALIVE; pkt->data[1] = KA_FLAG_SLEEP; + pkt->data[2] = 0xe8; pkt->data[3] = 3; pkt->data[4] = 5; pkt->data[6] = 200; // 1 s period, 0.5 s ACTIVE, 20 s SLEEP + CHECK(etcp_keepalive_on_recv(&sock, pkt, link, 11) == 1); + CHECK(link->ka_peer_sleeping && !link->keepalive_timer && link->rx_timeout_timer); + test_clock_run_for(inst.ua, 60000); CHECK(conn->links_up && !conn->down_timer); + struct NODE_CONN_DIRECT* handle = NULL; + CHECK(node_conn_direct_open(&inst, 42, NULL, NULL, &handle, NULL) == NCD_REUSED); + node_conn_direct_close(handle); CHECK(conn->fin_wait); + test_clock_run_for(inst.ua, 60000); CHECK(!deleted && conn->links_up && conn->fin_wait); + receive_control(conn, NCD_SUBCMD_KEEP_ALIVE); CHECK(!conn->fin_wait); + test_clock_run_for(inst.ua, 120000); CHECK(!conn->links_up && conn->down_timer && !deleted); + test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); CHECK(deleted == 1 && !inst.connections->count); + teardown(); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS sleeping peer deadline, FIN_WAIT and disappearance"); +} + +/* Мягкое закрытие работает без NCD entry; KEEP_ALIVE сохраняет транспорт дольше прежних 5 секунд. */ +static void check_graceful(void) { + setup(); struct ETCP_CONN* conn = ready_conn(0); + receive_control(conn, NCD_SUBCMD_CLOSE); CHECK(deleted == 1 && !inst.connections->count); teardown(); + setup(); conn = ready_conn(0); struct NODE_CONN_DIRECT* handle = NULL; + CHECK(node_conn_direct_open(&inst, 42, NULL, NULL, &handle, NULL) == NCD_REUSED); + receive_control(conn, NCD_SUBCMD_CLOSE); CHECK(!deleted && !conn->fin_wait); + CHECK(conn->send_input_q->count == 1); + struct ncd_control_msg* response = (void*)conn->send_input_q->head->dgram; + CHECK(response->subcmd == NCD_SUBCMD_KEEP_ALIVE); + node_conn_direct_close(handle); CHECK(conn->fin_wait); + receive_control(conn, NCD_SUBCMD_KEEP_ALIVE); CHECK(!conn->fin_wait); + test_clock_run_for(inst.ua, 70000); CHECK(!deleted && conn->links_up); + receive_control(conn, NCD_SUBCMD_CLOSE); CHECK(deleted == 1 && !inst.connections->count); teardown(); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS inbound CLOSE, owned refusal and durable KEEP_ALIVE"); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); test_clock_init(10000); + debug_set_category_level(DEBUG_CATEGORY_CONNECTION, DEBUG_LEVEL_DEBUG); + debug_set_category_level(DEBUG_CATEGORY_KEEPALIVE, DEBUG_LEVEL_DEBUG); + size_t baseline = u_get_allocated_count(); + check_pending(); CHECK(u_get_allocated_count() == baseline); + check_reconnect(); CHECK(u_get_allocated_count() == baseline); + check_owner(0); CHECK(u_get_allocated_count() == baseline); + check_owner(1); CHECK(u_get_allocated_count() == baseline); + check_silent_peer(0); CHECK(u_get_allocated_count() == baseline); + check_silent_peer(1); CHECK(u_get_allocated_count() == baseline); + check_sleep(); CHECK(u_get_allocated_count() == baseline); + check_graceful(); CHECK(u_get_allocated_count() == baseline); + test_clock_finish(); return 0; +}