utun2/tests/test_etcp_server_lifecycle.c

/* Входящий ETCP: реальные очереди, таймеры, NCD и управляемые часы без сетевого I/O. */
#include <stdlib.h>
#include <string.h>
#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) + PACKET_DATA_SIZE, "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;
}