diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c index b85389a8..ca32a76d 100644 --- a/src/transport_layer/etcp_connections.c +++ b/src/transport_layer/etcp_connections.c @@ -1287,7 +1287,8 @@ void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) { if (!link->mtu_remote) link->mtu_remote = link->mtu; etcp->tcp_link_count++; if (link->tcp_link) { const uint8_t* ed = stcp_link_get_peer_ed25519_pubkey(link->tcp_link); if (ed) memcpy(etcp->peer_ed25519_pubkey, ed, SC_PUBKEY_SIZE); } - if (!etcp->peer_reset_id) etcp_session_start(etcp); // сервер создаёт ETCP_CONN после STCP handshake + // Известная эпоха пира не подтверждает новый поток; следующий линк не перезапускает согласование. + if (!etcp->initialized && !etcp->session_required) etcp_session_start(etcp); if (etcp->initialized == 0) etcp_conn_ready(etcp); if (etcp->state == 2) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "enter_ready_tcp: conn DELETED after conn_ready, link=%p", (void*)link); return; } etcp_keepalive_link_ready(link); diff --git a/src/transport_layer/etcp_session.c b/src/transport_layer/etcp_session.c index cd51e223..d9e5d4e7 100644 --- a/src/transport_layer/etcp_session.c +++ b/src/transport_layer/etcp_session.c @@ -93,10 +93,8 @@ int etcp_session_receive(struct ETCP_CONN* c, const uint8_t* data, size_t len) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP,"[%s] session: stale target or empty cookie",c->log_name); } else if (data[0] == ETCP_SESSION_CHALLENGE) { session_send(c,ETCP_SESSION_CONFIRM,peer,cookie); - if (c->session_required || peer != c->peer_reset_id) { - etcp_session_observe(c,peer); - if (c->session_challenge) session_send(c,ETCP_SESSION_CHALLENGE,peer,c->session_challenge); - } + // Встречный challenge отправляет таймер: немедленный ответ зацикливает обмен при потере CONFIRM. + if (c->session_required || peer != c->peer_reset_id) etcp_session_observe(c,peer); } else if (peer == c->session_candidate && cookie == c->session_challenge) { etcp_session_cancel(c); if (peer != c->peer_reset_id) etcp_conn_reinit_id(c,"confirmed peer epoch",c->reset_id); diff --git a/tests/Makefile.am b/tests/Makefile.am index 0987f8ff..5e512b1e 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -326,9 +326,13 @@ test_transport_SOURCES = test_stcp_link.c test_transport_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib test_transport_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) -test_etcp_transport_SOURCES = test_etcp_stcp.c +test_etcp_transport_SOURCES = test_etcp_stcp.c $(test_clock_sources) test_etcp_transport_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib test_etcp_transport_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +test_etcp_transport_LDFLAGS = $(test_clock_ldflags) +if OS_LINUX +test_etcp_transport_LDFLAGS += -Wl,--wrap=etcp_encrypt_send +endif test_stcp_SOURCES = test_stcp.c test_stcp_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib diff --git a/tests/test_etcp_reconnect.c b/tests/test_etcp_reconnect.c index cbb1d97e..24416b4a 100644 --- a/tests/test_etcp_reconnect.c +++ b/tests/test_etcp_reconnect.c @@ -53,6 +53,7 @@ static int restart_action_done = 0; static int _dump_interval = 50; static int _dump_timer = 0; static uint8_t packet_buffer[PACKET_SIZE]; +static uint8_t received_sequences[8][TOTAL_PACKETS]; static int write_configs(void) { FILE* f = fopen(server_config_path, "w"); @@ -115,7 +116,12 @@ static int is_connection_established(struct UTUN_INSTANCE* inst) { struct ll_entry* entry = inst->connections->head; while (entry) { struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data; - if (ce->conn->initialized) return 1; + struct ETCP_CONN* conn = ce->conn; + struct UTUN_INSTANCE* peer = inst == client_instance ? server_instance : client_instance; + struct ETCP_CONN* remote = peer ? instance_find_conn(peer, inst->node_id) : NULL; + if (remote && conn->initialized && remote->initialized && !conn->session_required && !remote->session_required && + !conn->reinit_pending && !remote->reinit_pending && conn->peer_reset_id == remote->reset_id && + remote->peer_reset_id == conn->reset_id) return 1; entry = entry->next; } return 0; @@ -131,7 +137,18 @@ static void drain_received(int count_flag) { queue_set_callback(conn->output_queue, NULL, NULL); struct ETCP_FRAGMENT* pkt; while ((pkt = (struct ETCP_FRAGMENT*)queue_data_get(conn->output_queue)) != NULL) { - if (count_flag) packets_received++; + // output_queue содержит также служебные кодограммы; считаем только наши маркированные пакеты. + if (count_flag && pkt->ll.len >= 2 && pkt->ll.dgram[0] == 'T' && pkt->ll.dgram[1] == 'R') { + int valid = pkt->ll.len == PACKET_SIZE; + unsigned seq = valid ? pkt->ll.dgram[4] | ((unsigned)pkt->ll.dgram[5] << 8) : TOTAL_PACKETS; + if (valid) valid = pkt->ll.dgram[2] == phase && seq < TOTAL_PACKETS && !received_sequences[phase][seq]; + for (int i = 6; valid && i < PACKET_SIZE; i++) valid = pkt->ll.dgram[i] == (uint8_t)(seq + i + phase); + if (valid) { received_sequences[phase][seq] = 1; packets_received++; } + else { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "reconnect: invalid/duplicate payload phase=%d seq=%u len=%u", phase, seq, pkt->ll.len); + test_completed = 2; + } + } if (pkt->ll.dgram) memory_pool_free(conn->instance->data_pool, pkt->ll.dgram); queue_entry_free((struct ll_entry*)pkt); } @@ -146,8 +163,9 @@ static void send_packets(void) { if (!conn || !conn->input_queue) return; while (packets_sent < TOTAL_PACKETS) { if (queue_entry_count(conn->input_queue) >= MAX_QUEUE_SIZE) break; - packet_buffer[0] = (uint8_t)(packets_sent & 0xFF); - for (int i = 1; i < PACKET_SIZE; i++) packet_buffer[i] = (uint8_t)((packets_sent + i) % 256); + packet_buffer[0] = 'T'; packet_buffer[1] = 'R'; packet_buffer[2] = phase; packet_buffer[3] = 0; + packet_buffer[4] = packets_sent; packet_buffer[5] = packets_sent >> 8; + for (int i = 6; i < PACKET_SIZE; i++) packet_buffer[i] = (uint8_t)(packets_sent + i + phase); if (etcp_int_send(conn, packet_buffer, PACKET_SIZE) != 0) break; packets_sent++; } @@ -207,12 +225,10 @@ static void monitor(void* arg) { return; } printf("Server recreated (node_id=%llx)\n", (unsigned long long)server_instance->node_id); - // Force client to reinit — otherwise conn_ok stays true on old initialized=1 - { struct ll_entry* e = client_instance->connections->head; - while (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; etcp_conn_reinit(ce->conn, "test force reinit"); e = e->next; } } restart_action_done = 1; } drain_received(0); + conn_ok = is_connection_established(client_instance); if (conn_ok) { printf("=== Phase 3: Reconnected after server restart ===\n"); packets_sent = 0; @@ -228,6 +244,7 @@ static void monitor(void* arg) { last_reinit_count = cur_reinit; packets_sent = 0; packets_received = 0; + memset(received_sequences[phase], 0, sizeof(received_sequences[phase])); drain_received(0); } if (is_connection_established(client_instance)) @@ -291,6 +308,7 @@ static void monitor(void* arg) { last_reinit_count = cur_reinit; packets_sent = 0; packets_received = 0; + memset(received_sequences[phase], 0, sizeof(received_sequences[phase])); drain_received(0); } if (is_connection_established(client_instance)) diff --git a/tests/test_etcp_session.c b/tests/test_etcp_session.c index e295e1ee..a96de297 100644 --- a/tests/test_etcp_session.c +++ b/tests/test_etcp_session.c @@ -2,6 +2,7 @@ #include #include #include +#include #include "etcp.h" #include "etcp_api.h" #include "etcp_connections.h" @@ -171,6 +172,7 @@ static void teardown(void) { for (int i=0;i<2;i++) { for (int j=0;j<2;j++) { if (ep[i].links[j].rx_timeout_timer) uasync_cancel_timeout(ua, ep[i].links[j].rx_timeout_timer); + if (ep[i].links[j].keepalive_timer) uasync_cancel_timeout(ua, ep[i].links[j].keepalive_timer); } ep[i].c->links=NULL; // fixture owns synthetic authenticated links etcp_connection_close(ep[i].c); @@ -179,6 +181,107 @@ static void teardown(void) { for (int i=0;i<2;i++) { queue_free(ep[i].inst.connections); memory_pool_destroy(ep[i].inst.pkt_pool); } uasync_poll(ua,0); CHECK(ua->timer_alloc_count == ua->timer_free_count); uasync_destroy(ua,0); } + +// Пересоздаём настоящий ETCP_CONN с теми же ключами; линки/сокеты принадлежат fixture. +static void fresh_endpoint(int i) { + for (int j = 0; j < 2; j++) { + struct ETCP_LINK* link = &ep[i].links[j]; + if (link->keepalive_timer) { uasync_cancel_timeout(ua, link->keepalive_timer); link->keepalive_timer = NULL; } + if (link->rx_timeout_timer) { uasync_cancel_timeout(ua, link->rx_timeout_timer); link->rx_timeout_timer = NULL; } + } + ep[i].c->links = NULL; + etcp_connection_close(ep[i].c); uasync_poll(ua, 0); + ep[i].c = etcp_connection_create(&ep[i].inst, "fresh_session"); CHECK(ep[i].c); + ep[i].c->peer_node_id = 2 - i; + CHECK(sc_init_ctx(&ep[i].c->crypto_ctx, &ep[i].inst.my_keys) == SC_OK); + CHECK(sc_set_peer_public_key(&ep[i].c->crypto_ctx, ep[1-i].inst.my_keys.public_key, SC_PEER_PUBKEY_BIN) == SC_OK); + struct ll_entry* entry = ep[i].c->conn_queue_entry; + queue_remove_data(ep[i].c->conn_queue, entry); + ((struct conn_queue_entry*)entry->data)->peer_node_id = ep[i].c->peer_node_id; + CHECK(queue_data_put_with_index(ep[i].c->conn_queue, entry) == 0); + ep[i].c->links = &ep[i].links[0]; + for (int j = 0; j < 2; j++) ep[i].links[j].etcp = ep[i].c; + etcp_conn_add_cbk(ep[i].c, event, (void*)(intptr_t)i, ETCP_CBK_EVENT_REINIT | ETCP_CBK_EVENT_DOWN); + reinit[i] = down[i] = 0; +} + +static void transport_ready(int i, int j) { + struct ETCP_LINK* link = &ep[i].links[j]; + link->keepalive_interval = link->ka_period_ms = 1000; link->keepalive_timeout = 10000; + etcp_link_enter_ready_tcp(link); +} + +static void run_until(uint64_t end) { + while (get_time_tb() < end) pump(); +} + +static void initial_session_case(int scenario) { + setup(scenario == 4 || scenario == 5); + uint64_t local = ep[0].c->reset_id; + ep[0].c->next_tx_id = 73; ep[0].c->last_delivered_id = 42; ep[0].c->got_initial_pkt = 1; + if (scenario == 5) { + transport_ready(0, 1); + CHECK(!ep[0].c->session_required && !ep[0].c->session_timer && !reinit[0]); + CHECK(ep[0].c->reset_id == local && ep[0].c->next_tx_id == 73 && ep[0].c->last_delivered_id == 42); + } else { + if (scenario >= 1 && scenario <= 3) fresh_endpoint(0); + fresh_endpoint(1); + etcp_conn_apply_peer_reset_id(ep[1].c, ep[0].c->reset_id); + if (scenario >= 1 && scenario <= 3) { + if (scenario != 1) etcp_conn_apply_peer_reset_id(ep[0].c, ep[1].c->reset_id); + transport_ready(0, 0); + } + transport_ready(1, 0); + CHECK(ep[1].c->session_required && !ep[1].c->initialized); + if (scenario == 3 || scenario == 4) for (int i = 0; i < 2; i++) drop[i][2] = INT_MAX; + if (scenario == 3) { + unsigned before = tail; + run_until(get_time_tb() + 6000); + CHECK(ep[0].c->session_required && ep[1].c->session_required && drops); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Dropped CONFIRM: frames=%u dropped=%u interval=600ms", tail - before, drops); + CHECK(tail - before < 100); // Без таймера встречные CHALLENGE порождают тысячи ответов. + } else if (scenario == 4) { + for (unsigned k = 0; k < 100 && !ep[1].c->session_challenge; k++) pump(); + CHECK(ep[1].c->session_required && ep[1].c->session_challenge); + uint64_t cookie = ep[1].c->session_challenge, started = ep[1].c->session_started; + void* timer = ep[1].c->session_timer; + transport_ready(1, 1); + CHECK(ep[1].c->session_challenge == cookie && ep[1].c->session_started == started && ep[1].c->session_timer == timer); + } + memset(drop, 0, sizeof(drop)); converge(); + if (scenario == 0 || scenario == 4) { + CHECK(ep[0].c->reset_id == local && ep[0].c->next_tx_id == 1 && ep[0].c->last_delivered_id == 0); + CHECK(reinit[0] == 1 && reinit[1] == 0); + } + } + teardown(); + printf("PASS initial session scenario %d (restart, cold start, simultaneous, lost CONFIRM, pending/ready second link)\n", scenario); +} + +static void restart_during_confirmation(void) { + setup(0); + fresh_endpoint(1); etcp_conn_apply_peer_reset_id(ep[1].c, ep[0].c->reset_id); transport_ready(1, 0); + for (int i = 0; i < 2; i++) drop[i][2] = INT_MAX; + for (unsigned k = 0; k < 100 && !ep[0].c->session_challenge; k++) pump(); + CHECK(ep[0].c->session_challenge); + uint64_t old_peer = ep[0].c->peer_reset_id, candidate = ep[1].c->reset_id; + // Сохраняем правильный ответ первой попытки; после второго рестарта он должен стать stale. + struct ETCP_DGRAM* packet = memory_pool_alloc(ep[1].inst.pkt_pool); CHECK(packet); + packet->link = &ep[1].links[0]; packet->data_len = ETCP_SESSION_CONTROL_SIZE; packet->noencrypt_len = 0; + etcp_session_encode(packet->data, ETCP_SESSION_CONFIRM, candidate, ep[0].c->reset_id); + uint64_t cookie = htobe64(ep[0].c->session_challenge); memcpy(packet->data + 17, &cookie, sizeof(cookie)); + unsigned last = tail; + CHECK(etcp_encrypt_send(packet) > 0); memory_pool_free(ep[1].inst.pkt_pool, packet); + CHECK(tail == last + 1); struct frame stale = frames[last % 4096]; + fresh_endpoint(1); etcp_conn_apply_peer_reset_id(ep[1].c, ep[0].c->reset_id); transport_ready(1, 0); + CHECK(ep[1].c->reset_id != candidate); + for (unsigned k = 0; k < 100 && ep[0].c->session_candidate != ep[1].c->reset_id; k++) pump(); + CHECK(ep[0].c->session_candidate == ep[1].c->reset_id); + deliver(&stale, 0); CHECK(ep[0].c->peer_reset_id == old_peer && !reinit[0]); + memset(drop, 0, sizeof(drop)); converge(); CHECK(reinit[0] == 1); + teardown(); + puts("PASS peer restarts during confirmation; old valid CONFIRM cannot adopt the previous epoch"); +} static struct frame stream_packet(int data) { struct ETCP_DGRAM* p=memory_pool_alloc(ep[0].inst.pkt_pool); CHECK(p); p->link=&ep[0].links[0]; p->noencrypt_len=0; p->data_len=8; memset(p->data,0,8); p->data[0]=data ? 0 : 1; if (data) p->data[1]=99; @@ -263,6 +366,10 @@ int main(void) { timestamp_boundaries(); CHECK(u_get_allocated_count() == baseline); receive_window_case(0); CHECK(u_get_allocated_count() == baseline); receive_window_case(1); CHECK(u_get_allocated_count() == baseline); + for (int scenario = 0; scenario < 6; scenario++) { + initial_session_case(scenario); CHECK(u_get_allocated_count() == baseline); + } + restart_during_confirmation(); CHECK(u_get_allocated_count() == baseline); for (int scenario=0;scenario<9;scenario++) { setup(scenario == 5); saved[2]=stream_packet(0); saved[3]=stream_packet(1); @@ -272,10 +379,12 @@ int main(void) { if (scenario == 3) etcp_conn_fatal_reinit(ep[1].c,"simultaneous reset"); if (scenario == 4) { pump(); etcp_conn_fatal_reinit(ep[0].c,"second reset before confirmation"); } if (scenario == 6) { - blackhole=1; ep[0].c->session_started=get_time_tb()-100000; + blackhole=1; + uint64_t started = ep[0].c->session_started; CHECK(etcp_conn_ref_take(ep[0].c)==0); ep[0].c->links=NULL; // timeout closes a real connection, links belong to fixture - while (ep[0].c->state!=2) pump(); + while (ep[0].c->state!=2 && get_time_tb() < started + 110000) pump(); + CHECK(ep[0].c->state == 2 && get_time_tb() - started >= 100000); CHECK(!ep[0].c->session_timer && ep[0].c->delete_complete); etcp_conn_ref_free(ep[0].c); ep[0].c=NULL; ep[1].c->links=NULL; etcp_connection_close(ep[1].c); uasync_poll(ua,0); diff --git a/tests/test_etcp_stcp.c b/tests/test_etcp_stcp.c index 7224324e..377147b6 100644 --- a/tests/test_etcp_stcp.c +++ b/tests/test_etcp_stcp.c @@ -1,162 +1,267 @@ -// test_etcp_stcp.c — integration: etcp_send + etcp_bind over STCP link -#include "stcp_link.h" -#include "etcp_api.h" +// Реальные STCP/REALITY-линки: встречное подключение и рестарт пира без reset выжившего узла. +#include +#include +#include #include "etcp.h" +#include "etcp_api.h" #include "etcp_connections.h" -#include "secure_channel.h" +#include "etcp_session.h" +#include "node_conn_direct.h" #include "topo_group.h" #include "topo_node.h" -#include "node_conn_direct.h" -#include "../src/utun_instance.h" -#include "../lib/u_async.h" -#include "../lib/ll_queue.h" +#include "utun_instance.h" #include "../lib/debug_config.h" #include "../lib/mem.h" -#include -#include -#include -#include +#include "test_clock.h" -static int test_failed = 0; -static struct SC_MYKEYS s_keys, c_keys; -static struct UTUN_INSTANCE *srv_inst, *cli_inst; -static struct ETCP_CONN *srv_conn, *cli_conn; -static int g_srv_recv_count = 0; -static uint8_t g_srv_recv_data[256]; +#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "FAIL %s:%d: %s", __func__, __LINE__, #x); exit(1); } } while (0) +#define TEST_CMD 253 +#define MESSAGE_COUNT 16 +#define FULL_MASK ((1u << MESSAGE_COUNT) - 1) -#define TASSERT(cond) do { \ - if (!(cond)) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, " FAIL: %s", #cond); test_failed = 1; return test_failed; } \ -} while(0) +struct endpoint { + struct UTUN_INSTANCE* inst; + struct NODE_CONN_DIRECT* owner; + struct ETCP_CONN* conn; + uint64_t id; + uint16_t port; + unsigned received, reinit, deleted; +}; +struct message { uint8_t cmd, from, phase, seq, body[64]; }; +static struct endpoint endpoints[2]; +static struct UASYNC* ua; +static unsigned phase; +static int failed; +static uint64_t stream_packets, session_packets; +static const char* public_keys[2] = { + "b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01", + "ce8871f07fa056c636d297115f231b08c29cdf94e0d440fce83a07c34416d36a" +}; +static const char* private_keys[2] = { + "704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f", + "38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68" +}; -static void etcp_recv_cb(struct ETCP_CONN *conn, struct ll_entry *entry) { +#ifdef __linux__ +// Считаем реальные отправки потока/control: тест ловит шторм независимо от скорости CPU. +int __real_etcp_encrypt_send(struct ETCP_DGRAM* packet); +int __wrap_etcp_encrypt_send(struct ETCP_DGRAM* packet) { + if (packet && packet->data_len) { + if (packet->data[0] <= ETCP_SECTION_ACK) stream_packets++; + if (packet->data[0] >= ETCP_SESSION_HELLO && packet->data[0] <= ETCP_SESSION_CONFIRM) session_packets++; + } + return __real_etcp_encrypt_send(packet); +} +#endif + +static void fill_message(struct message* message, int from, unsigned seq) { + message->cmd = TEST_CMD; message->from = from; message->phase = phase; message->seq = seq; + for (unsigned i = 0; i < sizeof(message->body); i++) message->body[i] = (phase * 17 + seq * 3 + i) ^ from; +} + +static void receive_message(struct ETCP_CONN* conn, struct ll_entry* entry) { + int to = conn->instance == endpoints[0].inst ? 0 : 1; + struct message expected; + int valid = entry->len == sizeof(expected) && entry->dgram; + const struct message* received = valid ? (const struct message*)entry->dgram : NULL; + if (valid) valid = received->phase == phase && received->from == 1 - to && received->seq < MESSAGE_COUNT; + if (valid) { + fill_message(&expected, 1 - to, received->seq); + valid = !memcmp(received, &expected, sizeof(expected)) && !(endpoints[to].received & (1u << received->seq)); + } + if (valid) endpoints[to].received |= 1u << received->seq; + else { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Invalid/duplicate message: to=%d phase=%u bytes=%u", to, phase, entry->len); failed = 1; } + queue_dgram_free(entry); queue_entry_free(entry); +} + +static void connection_event(struct ETCP_CONN* conn, int event, void* arg) { (void)conn; - g_srv_recv_count++; - if (entry->dgram && entry->len) { - size_t n = entry->len < 256 ? entry->len : 255; - memcpy(g_srv_recv_data, entry->dgram, n); + struct endpoint* endpoint = arg; + if (event == ETCP_CBK_EVENT_REINIT) endpoint->reinit++; + if (event == ETCP_CBK_EVENT_DELETE) endpoint->deleted++; +} + +static void connect_done(void* arg, struct ETCP_CONN* conn, int type) { + struct endpoint* endpoint = arg; + if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Connection failed: local=%016llx", (unsigned long long)endpoint->id); failed = 1; } + else if (type & ETCP_CONNECT_EARLY) endpoint->conn = conn; +} + +static void start_endpoint(int i, int reality) { + struct endpoint* endpoint = &endpoints[i]; + struct SC_MYKEYS keys; + CHECK(sc_init_local_keys(&keys, public_keys[i], private_keys[i]) == SC_OK); + uint64_t id = sc_derive_node_id_from_pubkey(keys.public_key); + char config[2048], reality_config[512] = ""; + if (reality) snprintf(reality_config, sizeof(reality_config), + "[reality]\nenabled=1\nprivate_key=%s\nshort_ids=0102030405060708\nserver_name=localhost.test\n" + "dest=127.0.0.1:1\nfingerprint=openssl\n", private_keys[i]); + snprintf(config, sizeof(config), + "[global]\nmy_node_id=0x%016llx\nmy_node_name=transport-%d\nmy_public_key=%s\nmy_private_key=%s\n" + "[server:stcp]\naddr=127.0.0.1:%u\ntype=public\ntransport=tcp\nreality=%d\n%s", + (unsigned long long)id, i, public_keys[i], private_keys[i], endpoint->port, reality, reality_config); + endpoint->inst = utun_instance_create_from_str(ua, config); CHECK(endpoint->inst); + CHECK(utun_instance_init(endpoint->inst) >= 0); + CHECK(etcp_bind(endpoint->inst, TEST_CMD, receive_message) == 0); + endpoint->id = endpoint->inst->node_id; + struct ETCP_SOCKET* socket = endpoint->inst->etcp_sockets; + while (socket && !socket->is_tcp) socket = socket->next; + CHECK(socket && socket->local_addr.ss_family == AF_INET); + endpoint->port = ntohs(((struct sockaddr_in*)&socket->local_addr)->sin_port); CHECK(endpoint->port); + endpoint->conn = NULL; endpoint->owner = NULL; endpoint->received = endpoint->reinit = endpoint->deleted = 0; +} + +static void stop_endpoint(int i) { + struct endpoint* endpoint = &endpoints[i]; + if (endpoint->owner) { node_conn_direct_close(endpoint->owner); endpoint->owner = NULL; } + endpoint->inst->running = 0; + utun_instance_destroy(endpoint->inst); endpoint->inst = NULL; endpoint->conn = NULL; +} + +static struct TOPO_NODE* import_peer(int to) { + struct UTUN_INSTANCE* from = endpoints[1-to].inst; + struct TOPO_GROUP_NODE* local = topo_groups_get_default(from->topo_groups)->local_node; + struct TOPO_NODE* info = topo_node_registry_find(from->topo_groups, from->node_id); CHECK(info); + uint8_t wire[4096]; int len = topo_node_serialize(info, local, 0, 0, wire, sizeof(wire), 0); CHECK(len > 0); + struct TOPO_NODE* imported = NULL; + CHECK(topo_node_deserialize(topo_groups_get_default(endpoints[to].inst->topo_groups), wire, len, + &imported, NULL, NULL, NULL, NULL) == 0); + CHECK(topo_node_registry_store(endpoints[to].inst->topo_groups, imported)); + return imported; +} + +static int pair_ready(void) { + for (int i = 0; i < 2; i++) { + endpoints[i].conn = instance_find_conn(endpoints[i].inst, endpoints[1-i].id); + struct ETCP_CONN* conn = endpoints[i].conn; + if (!conn || !conn->initialized || !conn->links_up || conn->session_required || conn->reinit_pending) return 0; } - queue_dgram_free(entry); - queue_entry_free(entry); + return endpoints[0].conn->peer_reset_id == endpoints[1].conn->reset_id && + endpoints[1].conn->peer_reset_id == endpoints[0].conn->reset_id; } -static void connect_cb(void *arg, struct ETCP_CONN *conn, int type) { - (void)arg; - if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "etcp_connect failed"); test_failed = 1; return; } - if (type & ETCP_CONNECT_EARLY) { - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "etcp_connect: EARLY ready"); - cli_conn = conn; +static void wait_pair(void) { + uint64_t end = get_time_tb() + 50000; + while (!failed && get_time_tb() < end) { + test_clock_poll(ua, 100); + if (pair_ready()) return; } + DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Session recovery timed out: phase=%u stream=%llu control=%llu", phase, + (unsigned long long)stream_packets, (unsigned long long)session_packets); + CHECK(0); } -static char *build_server_config(int port) { - static char buf[512]; - snprintf(buf, sizeof(buf), - "[global]\n" - "my_node_id=0xAAAAAAAAAAAAAAAA\n" - "my_private_key=38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68\n" - "my_public_key=ce8871f07fa056c636d297115f231b08c29cdf94e0d440fce83a07c34416d36a\n" - "\n" - "[server: stcp]\n" - "addr=127.0.0.1:%d\n" - "type=public\n" - "transport=tcp\n", - port); - return buf; +static void check_unique(int i) { + unsigned count = 0; + for (struct ll_entry* e = endpoints[i].inst->connections->head; e; e = e->next) + if (((struct conn_queue_entry*)e->data)->peer_node_id == endpoints[1-i].id) count++; + CHECK(count == 1); } -static char *build_client_config(int port) { - static char buf[512]; - snprintf(buf, sizeof(buf), - "[global]\n" - "my_node_id=0xBBBBBBBBBBBBBBBB\n" - "my_private_key=704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f\n" - "my_public_key=b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01\n" - "\n" - "[server: stcp]\n" - "addr=127.0.0.1:%d\n" - "type=public\n" - "transport=tcp\n", - port); - return buf; +static void hold_connection(int i) { + struct endpoint* endpoint = &endpoints[i]; + struct TOPO_NODE* peer = topo_node_registry_find(endpoint->inst->topo_groups, endpoints[1-i].id); CHECK(peer); + CHECK(node_conn_direct_open_node(endpoint->inst, peer->node_id, NULL, NULL, &endpoint->owner, peer, NULL) == NCD_REUSED); + CHECK(node_conn_direct_get_conn(endpoint->owner) == endpoint->conn); + etcp_conn_add_cbk(endpoint->conn, connection_event, endpoint, ETCP_CBK_EVENT_REINIT | ETCP_CBK_EVENT_DELETE); } -int main(void) { - debug_config_init(); - debug_set_level(DEBUG_LEVEL_INFO); - debug_set_categories(DEBUG_CATEGORY_GENERAL | DEBUG_CATEGORY_SOCKET | DEBUG_CATEGORY_CRYPTO); - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "=== etcp_send/bind over STCP link ==="); - - srand((unsigned)time(NULL)); - struct UASYNC *ua = uasync_create(); TASSERT(ua); - utun_instance_set_tun_init_enabled(0); - - srv_inst = utun_instance_create_from_str(ua, build_server_config(0)); - TASSERT(srv_inst); - TASSERT(utun_instance_init(srv_inst) >= 0); - - cli_inst = utun_instance_create_from_str(ua, build_client_config(0)); - TASSERT(cli_inst); - TASSERT(utun_instance_init(cli_inst) >= 0); - - TASSERT(etcp_bind(srv_inst, 0, etcp_recv_cb) >= 0); - - struct TOPO_GROUP_NODE *srv_node = topo_groups_get_default(srv_inst->topo_groups) ? topo_groups_get_default(srv_inst->topo_groups)->local_node : NULL; - TASSERT(srv_node); - { struct TOPO_NODE* srv_ni = topo_node_registry_find(srv_inst->topo_groups, srv_node->node_id); TASSERT(srv_ni); - uint8_t wire[4096]; - int len = topo_node_serialize(srv_ni, srv_node, 0, 0, wire, sizeof(wire), 0); - TASSERT(len > 0); - struct TOPO_NODE* cli_ni = NULL; - struct TOPO_GROUP* cli_group = topo_groups_get_default(cli_inst->topo_groups); - TASSERT(topo_node_deserialize(cli_group, wire, (size_t)len, &cli_ni, NULL, NULL, NULL, NULL) == 0); - TASSERT(topo_node_registry_store(cli_inst->topo_groups, cli_ni)); } - cli_inst->etcp_connect_timeout_tb = 100000; - TASSERT(etcp_connect(cli_inst, srv_node, connect_cb, NULL, ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE) == 0); - - int ticks = 0; - while (!cli_conn && ticks < 5000) { uasync_poll(ua, 10); ticks++; } - TASSERT(cli_conn); - int initial_links = 0; - for (struct ETCP_LINK* link = cli_conn->links; link; link = link->next) initial_links++; - struct NODE_CONN_DIRECT *owner1 = NULL, *owner2 = NULL; - struct TOPO_NODE* peer = topo_node_registry_find(cli_inst->topo_groups, srv_node->node_id); - TASSERT(node_conn_direct_open_node(cli_inst, peer->node_id, NULL, NULL, &owner1, peer, NULL) == NCD_REUSED); - TASSERT(node_conn_direct_open_node(cli_inst, peer->node_id, NULL, NULL, &owner2, peer, NULL) == NCD_REUSED); - TASSERT(node_conn_direct_update_node(cli_inst, peer->node_id) == 0); - int repeated_links = 0; - for (struct ETCP_LINK* link = cli_conn->links; link; link = link->next) repeated_links++; - TASSERT(repeated_links == initial_links); - node_conn_direct_close(owner1); - TASSERT(node_conn_direct_get_conn(owner2) == cli_conn); - - { struct ll_entry* e = srv_inst->connections->head; - while (e) { struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; - struct ETCP_CONN *c = ce->conn; - struct ETCP_LINK *l = c->links; while (l) { if (l->is_tcp) { srv_conn = c; break; } l = l->next; } - if (srv_conn) break; - e = e->next; } +static void exchange_messages(void) { + for (int i = 0; i < 2; i++) endpoints[i].received = 0; + for (int i = 0; i < 2; i++) for (unsigned seq = 0; seq < MESSAGE_COUNT; seq++) { + struct ll_entry* entry = ll_alloc_lldgram(sizeof(struct message)); CHECK(entry); + fill_message((struct message*)entry->dgram, i, seq); entry->len = sizeof(struct message); + CHECK(etcp_send(endpoints[i].conn, entry) == 0); + } + uint64_t end = get_time_tb() + 30000; + while (!failed && get_time_tb() < end) { + test_clock_poll(ua, 100); + if (endpoints[0].received != FULL_MASK || endpoints[1].received != FULL_MASK) continue; + int drained = 1; + for (int i = 0; i < 2; i++) { + struct ETCP_CONN* conn = endpoints[i].conn; + if (conn->input_wait_ack->count || conn->input_send_q->count || conn->input_queue->count || conn->send_input_q->count) drained = 0; + } + if (drained) break; } - TASSERT(srv_conn); - - const char *msg = "hello via etcp_send!"; - struct ll_entry *e = queue_entry_new(0); - e->dgram = u_malloc(strlen(msg) + 1); - strcpy((char *)e->dgram, msg); - e->len = (uint16_t)strlen(msg); - TASSERT(etcp_send(cli_conn, e) == 0); - - ticks = 0; - while (g_srv_recv_count < 1 && ticks < 2000) { uasync_poll(ua, 10); ticks++; } - TASSERT(g_srv_recv_count == 1); - TASSERT(memcmp(g_srv_recv_data, msg, strlen(msg)) == 0); - node_conn_direct_close(owner2); - - srv_inst->running = 0; cli_inst->running = 0; - utun_instance_destroy(srv_inst); srv_inst = NULL; - utun_instance_destroy(cli_inst); cli_inst = NULL; + CHECK(!failed && endpoints[0].received == FULL_MASK && endpoints[1].received == FULL_MASK); + for (int i = 0; i < 2; i++) CHECK(!endpoints[i].conn->input_wait_ack->count && !endpoints[i].conn->input_send_q->count); +} + +static void run_case(int reality, int simultaneous, int restarted) { + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Transport case: reality=%d simultaneous=%d restart=%d", reality, simultaneous, restarted); + memset(endpoints, 0, sizeof(endpoints)); phase = 0; failed = 0; stream_packets = session_packets = 0; + ua = uasync_create(); CHECK(ua); + start_endpoint(0, reality); start_endpoint(1, reality); + struct TOPO_NODE* peers[2] = {import_peer(0), import_peer(1)}; + for (int i = 0; i <= simultaneous; i++) { + struct TOPO_GROUP_NODE node = {0}; node.node_id = peers[i]->node_id; + CHECK(etcp_connect(endpoints[i].inst, &node, connect_done, &endpoints[i], ETCP_CONNECT_EARLY | ETCP_CONNECT_LATE) == 0); + } + wait_pair(); + for (int i = 0; i < 2; i++) { check_unique(i); hold_connection(i); } + // Повторные владельцы и обновление адресов не создают дополнительные линки. + unsigned links = 0; for (struct ETCP_LINK* l = endpoints[0].conn->links; l; l = l->next) links++; + struct NODE_CONN_DIRECT* extra = NULL; + CHECK(node_conn_direct_open_node(endpoints[0].inst, peers[0]->node_id, NULL, NULL, &extra, peers[0], NULL) == NCD_REUSED); + CHECK(node_conn_direct_update_node(endpoints[0].inst, peers[0]->node_id) == 0); + unsigned updated = 0; for (struct ETCP_LINK* l = endpoints[0].conn->links; l; l = l->next) updated++; + CHECK(updated == links); node_conn_direct_close(extra); + exchange_messages(); + + if (restarted < 2) { + int survivor = 1 - restarted; + struct ETCP_CONN* retained = endpoints[survivor].conn; + CHECK(etcp_conn_ref_take(retained) == 0); + uint64_t local_epoch = retained->reset_id, peer_epoch = retained->peer_reset_id; + unsigned before = endpoints[survivor].reinit; + uint64_t packets_before = stream_packets; + stop_endpoint(restarted); start_endpoint(restarted, reality); + phase++; wait_pair(); + CHECK(!endpoints[survivor].deleted && endpoints[survivor].conn == retained); + CHECK(retained->reset_id == local_epoch && retained->peer_reset_id != peer_epoch); + CHECK(endpoints[survivor].reinit == before + 1); + CHECK(node_conn_direct_get_conn(endpoints[survivor].owner) == retained); +#ifdef __linux__ + CHECK(stream_packets - packets_before < 256); +#endif + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Recovery complete: stream=%llu control=%llu REINIT=1 survivor_epoch=%016llx", + (unsigned long long)(stream_packets - packets_before), (unsigned long long)session_packets, + (unsigned long long)local_epoch); + etcp_conn_ref_free(retained); + check_unique(0); check_unique(1); exchange_messages(); + } else { + uint64_t old_epochs[2] = {endpoints[0].conn->reset_id, endpoints[1].conn->reset_id}; + stop_endpoint(0); stop_endpoint(1); start_endpoint(0, reality); start_endpoint(1, reality); + peers[0] = import_peer(0); peers[1] = import_peer(1); + for (int i = 0; i < 2; i++) { + struct TOPO_GROUP_NODE node = {0}; node.node_id = peers[i]->node_id; + CHECK(etcp_connect(endpoints[i].inst, &node, connect_done, &endpoints[i], ETCP_CONNECT_EARLY) == 0); + } + phase++; wait_pair(); + for (int i = 0; i < 2; i++) { CHECK(endpoints[i].conn->reset_id != old_epochs[i]); check_unique(i); } + exchange_messages(); + } + stop_endpoint(0); stop_endpoint(1); + uasync_poll(ua, 0); + CHECK(ua->timer_alloc_count == ua->timer_free_count && ua->socket_alloc_count == ua->socket_free_count); uasync_destroy(ua, 0); + printf("PASS transport reality=%d simultaneous=%d restart=%d: epochs, bidirectional delivery and resource balance\n", + reality, simultaneous, restarted); +} - if (test_failed) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "=== FAILED ==="); return 1; } - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "=== PASSED ==="); +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); + if (getenv("UTUN_TEST_DEBUG")) { + debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_DEBUG); + debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_INFO); + } + test_clock_init(100); utun_instance_set_tun_init_enabled(0); + for (int reality = 0; reality < 2; reality++) { + for (int simultaneous = 0; simultaneous < 2; simultaneous++) + for (int restarted = 0; restarted < 2; restarted++) run_case(reality, simultaneous, restarted); + run_case(reality, 1, 2); + } + test_clock_finish(); return 0; }