diff --git a/src/control_server.c b/src/control_server.c index 815641d7..be5a858a 100644 --- a/src/control_server.c +++ b/src/control_server.c @@ -1389,7 +1389,7 @@ static struct ETCP_CONN* find_connection_by_peer_id(struct UTUN_INSTANCE* instan static void send_single_node_info(struct control_server* server, struct control_client* client, struct TOPO_NODEQ* nq) { if (!server || !nq || !nq->node) return; uint8_t buf[8192]; - int ser_len = topo_node_serialize(topo_groups_get_default(server->instance->topo_groups), nq, buf + 2, sizeof(buf) - 2); + int ser_len = topo_node_serialize(topo_groups_get_default(server->instance->topo_groups), nq, buf + 2, sizeof(buf) - 2, 0); if (ser_len < 0) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to serialize node info"); return; } buf[0] = CTRL_BGP_CMD; buf[1] = CTRL_BGP_SUBCMD; size_t data_size = 2 + (size_t)ser_len; diff --git a/src/routing_layer/conn_mgr.c b/src/routing_layer/conn_mgr.c index 6454a16f..1fd68a55 100644 --- a/src/routing_layer/conn_mgr.c +++ b/src/routing_layer/conn_mgr.c @@ -18,6 +18,7 @@ #include "config_parser.h" #include "secure_channel.h" #include "topo_node.h" +#include "topo_node_sqlite.h" #include "topo_group.h" #include "route_ping.h" #include "route_connectivity.h" @@ -70,20 +71,22 @@ struct cm_exchange_pending { uint8_t probes_done; }; -struct CONN_MGR* conn_mgr_init(struct UTUN_INSTANCE* instance) { - if (!instance) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "NULL instance"); return NULL; } +struct CONN_MGR* conn_mgr_init(struct TOPO_GROUP* group) { + if (!group || !group->instance) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "NULL group"); return NULL; } struct CONN_MGR* mgr = u_calloc(1, sizeof(struct CONN_MGR)); if (!mgr) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "alloc failed"); return NULL; } - mgr->instance = instance; + mgr->instance = group->instance; + mgr->group = group; mgr->entry_capacity = 8; mgr->entries = u_calloc(mgr->entry_capacity, sizeof(struct CONN_MGR_ENTRY)); if (!mgr->entries) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "entries alloc failed"); u_free(mgr); return NULL; } - etcp_router_bind(instance, ETCP_RT_ID_CONN_MGR, conn_mgr_router_recv_handler); - mgr->bg_ping_timer = uasync_set_timeout(instance->ua, CONN_MGR_BG_PING_INTERVAL_TB, mgr, cm_bg_ping_timer_cb, "conn_mgr_bg_ping"); - mgr->candidate_ping_timer = uasync_set_timeout(instance->ua, CONN_MGR_CANDIDATE_PING_TB, mgr, cm_candidate_ping_timer_cb, "conn_mgr_cand_ping"); + mgr->direct_timeout_ms = CONN_MGR_CONNECT_DIRECT_TIMEOUT_MS; + etcp_router_bind(mgr->instance, ETCP_RT_ID_CONN_MGR, conn_mgr_router_recv_handler); + mgr->bg_ping_timer = uasync_set_timeout(mgr->instance->ua, CONN_MGR_BG_PING_INTERVAL_TB, mgr, cm_bg_ping_timer_cb, "conn_mgr_bg_ping"); + mgr->candidate_ping_timer = uasync_set_timeout(mgr->instance->ua, CONN_MGR_CANDIDATE_PING_TB, mgr, cm_candidate_ping_timer_cb, "conn_mgr_cand_ping"); mgr->bg_ping_cycle_start_tb = get_time_tb(); mgr->initialized = 1; - DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: initialized, bg_ping started"); + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: initialized for group %016llx, bg_ping started", (unsigned long long)group->group_id); return mgr; } @@ -157,7 +160,7 @@ static void cm_deliver_result(struct CONN_MGR_ENTRY* entry, int result) { } static void cm_update_nodeinfo(struct CONN_MGR_ENTRY* entry) { - struct TOPO_GROUP* group = topo_groups_get_default(entry->mgr->instance->topo_groups); + struct TOPO_GROUP* group = entry->mgr->group; if (!group) return; struct TOPO_NODEQ* nq = topo_node_find_by_id(group, entry->node_id); if (!nq) return; @@ -167,18 +170,51 @@ static void cm_update_nodeinfo(struct CONN_MGR_ENTRY* entry) { } static void cm_clear_nodeinfo(struct CONN_MGR* mgr, uint64_t node_id) { - struct TOPO_NODEQ* nq = topo_node_find_by_id(topo_groups_get_default(mgr->instance->topo_groups), node_id); + struct TOPO_NODEQ* nq = topo_node_find_by_id(mgr->group, node_id); if (!nq) return; nq->conn_mgr_type = CONN_TYPE_NONE; nq->conn_mgr_intermediariy_count = 0; } +static void cm_cleanup_db_node(struct CONN_MGR_ENTRY* entry) { + if (!entry->db_loaded) return; + struct TOPO_GROUP* group = entry->mgr->group; + struct TOPO_NODEQ* nq = topo_node_find_by_id(group, entry->node_id); + if (!nq) return; + if (nq->paths && queue_entry_count(nq->paths) > 0) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: db_node 0x%016llx got paths, keep in group", (unsigned long long)entry->node_id); + return; + } + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: removing db_node 0x%016llx from group (no paths)", (unsigned long long)entry->node_id); + queue_remove_data(group->nodes, &nq->ll); + topo_node_registry_release(group->instance->topo_groups, nq->node); + topo_node_free_lists(group, nq); + queue_entry_free(&nq->ll); +} + int conn_mgr_connect_node(struct CONN_MGR* mgr, uint64_t node_id, uint32_t idle_timeout_ms, conn_mgr_connect_callback_t cb, void* cb_arg) { if (!mgr) return CONN_MGR_ERR_INTERNAL; - struct TOPO_GROUP* group = topo_groups_get_default(mgr->instance->topo_groups); + struct TOPO_GROUP* group = mgr->group; if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "BGP not initialized"); return CONN_MGR_ERR_INTERNAL; } struct TOPO_NODEQ* target = topo_node_find_by_id(group, node_id); + int loaded_from_db = 0; + if (!target && mgr->instance->topo_sqlite_db && mgr->instance->topo_groups) { + struct TOPO_NODE* ni = topo_node_sqlite_node_load(mgr->instance->topo_sqlite_db, mgr->instance->topo_groups, node_id); + if (ni) { + ni = topo_node_registry_acquire(mgr->instance->topo_groups, ni); + struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_NODEQ)); + if (qe) { + target = (struct TOPO_NODEQ*)qe; + memset((uint8_t*)target + sizeof(struct ll_entry), 0, sizeof(*target) - sizeof(struct ll_entry)); + target->node = ni; target->hash_node_id = node_id; + target->conn_mgr_type = CONN_TYPE_NONE; + queue_data_put_with_index(group->nodes, &target->ll); + loaded_from_db = 1; + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: node 0x%016llx loaded from DB", (unsigned long long)node_id); + } else { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: failed to alloc TOPO_NODEQ for DB node 0x%016llx", (unsigned long long)node_id); } + } + } if (!target) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "node 0x%016llx not found", (unsigned long long)node_id); return CONN_MGR_ERR_NOT_FOUND; } struct CONN_MGR_ENTRY* entry = cm_find_entry(mgr, node_id); if (entry && entry->state == CONN_MGR_STATE_CONNECTED) { @@ -221,6 +257,7 @@ int conn_mgr_connect_node(struct CONN_MGR* mgr, uint64_t node_id, uint32_t idle_ entry->main.phase = 0; entry->main.timer = NULL; entry->main.request_id = 0; + entry->db_loaded = (uint8_t)loaded_from_db; int has_our_local = cm_has_local_addr(group->local_node); int has_target_local = cm_has_local_addr(target); if (has_our_local && has_target_local) { entry->local_scan_state = CM_TRY_PENDING; cm_start_local_scan(entry); } @@ -238,7 +275,7 @@ int conn_mgr_disconnect_node(struct CONN_MGR* mgr, uint64_t node_id) { pkt.subcmd = CONN_MGR_SUBCMD_DISCONNECT; pkt.node_id = node_id; struct ll_entry* qe = queue_entry_new(0); - if (qe) { qe->dgram = u_malloc(sizeof(pkt)); memcpy(qe->dgram, &pkt, sizeof(pkt)); qe->len = sizeof(pkt); etcp_route_send(mgr->instance, TOPO_GROUP_UTUN, node_id, qe, 1); } + if (qe) { qe->dgram = u_malloc(sizeof(pkt)); memcpy(qe->dgram, &pkt, sizeof(pkt)); qe->len = sizeof(pkt); etcp_route_send(mgr->instance, mgr->group->group_id, node_id, qe, 1); } cm_clear_nodeinfo(mgr, node_id); cm_entry_destroy(entry); entry->state = CONN_MGR_STATE_DISCONNECTED; @@ -258,6 +295,10 @@ int conn_mgr_set_idle_timeout(struct CONN_MGR* mgr, uint64_t node_id, uint32_t t return CONN_MGR_OK; } +void conn_mgr_set_direct_timeout_ms(struct CONN_MGR* mgr, uint32_t timeout_ms) { + if (mgr) mgr->direct_timeout_ms = timeout_ms; +} + int conn_mgr_get_status(struct CONN_MGR* mgr, uint64_t node_id, uint8_t* out_state, uint8_t* out_conn_type) { if (!mgr || !out_state || !out_conn_type) return CONN_MGR_ERR_INTERNAL; struct CONN_MGR_ENTRY* entry = cm_find_entry(mgr, node_id); @@ -272,7 +313,7 @@ int conn_mgr_send(struct CONN_MGR* mgr, uint64_t node_id, struct ll_entry* entry struct CONN_MGR_ENTRY* e = cm_find_entry(mgr, node_id); if (!e || e->state != CONN_MGR_STATE_CONNECTED) { queue_entry_free(entry); return -1; } e->last_traffic_tb = get_time_tb(); - struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(mgr->instance, TOPO_GROUP_UTUN, node_id, ETCP_RT_ID_CONN_MGR); + struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(mgr->instance, mgr->group->group_id, node_id, ETCP_RT_ID_CONN_MGR); if (!rconn) { queue_entry_free(entry); return -1; } int ret = etcp_router_conn_send(rconn, entry->dgram, entry->len); queue_entry_free(entry); @@ -401,7 +442,7 @@ struct cm_ping_ctx { }; static void cm_start_local_scan(struct CONN_MGR_ENTRY* entry) { - struct TOPO_GROUP* group = topo_groups_get_default(entry->mgr->instance->topo_groups); + struct TOPO_GROUP* group = entry->mgr->group; struct TOPO_NODEQ* target = topo_node_find_by_id(group, entry->node_id); if (!target || !target->node) { entry->local_scan_state = CM_TRY_FAILED; return; } for (const struct TOPO_ADDR4* a = target->node->v4_addrs; a; a = a->next) { @@ -429,9 +470,39 @@ static void cm_start_local_scan(struct CONN_MGR_ENTRY* entry) { static void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) { if (entry->main_connect_state == CM_TRY_OK) return; if (entry->local_scan_state == CM_TRY_OK) return; - struct TOPO_GROUP* group = topo_groups_get_default(entry->mgr->instance->topo_groups); + struct TOPO_GROUP* group = entry->mgr->group; struct TOPO_NODEQ* target = topo_node_find_by_id(group, entry->node_id); if (!target || !target->node) { cm_deliver_result(entry, CONN_MGR_ERR_NOT_FOUND); return; } + + if (entry->db_loaded) { + for (const struct TOPO_ADDR4* a = target->node->v4_addrs; a; a = a->next) { + if (a->protocol != TOPO_PROTO_UDP) continue; + struct ETCP_SOCKET* s = entry->mgr->instance->etcp_sockets; + while (s) { + if (s->local_addr.ss_family == AF_INET) { + struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET; + memcpy(&sin.sin_addr.s_addr, a->addr, 4); sin.sin_port = htons(a->port); + struct sockaddr_storage sa; memset(&sa, 0, sizeof(sa)); memcpy(&sa, &sin, sizeof(sin)); + struct ETCP_CONN* conn = etcp_connection_create(entry->mgr->instance, NULL); + if (conn) { + etcp_conn_add_init_cbk(conn, cm_direct_init_cb, entry); + sc_init_ctx(&conn->crypto_ctx, &entry->mgr->instance->my_keys); + sc_set_peer_public_key(&conn->crypto_ctx, target->node->public_key, 0); + if (etcp_link_new(conn, s, &sa, 0)) { + entry->main.timer = uasync_set_timeout(entry->mgr->instance->ua, (int)(entry->mgr->direct_timeout_ms * 10), entry, cm_direct_timeout_cb, "conn_mgr_db_direct"); + return; + } + } + } + s = s->next; + } + } + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: db_node 0x%016llx direct failed (no compatible addr/socket)", (unsigned long long)entry->node_id); + cm_cleanup_db_node(entry); + cm_deliver_result(entry, CONN_MGR_ERR_UNREACHABLE); + return; + } + uint8_t priority_order[] = { TOPO_ADDR_NAT, TOPO_ADDR_INTERFACE }; for (int pri = 0; pri < 2; pri++) { for (const struct TOPO_ADDR4* a = target->node->v4_addrs; a; a = a->next) { @@ -450,7 +521,7 @@ static void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) { sc_init_ctx(&conn->crypto_ctx, &entry->mgr->instance->my_keys); sc_set_peer_public_key(&conn->crypto_ctx, target->node->public_key, 0); if (etcp_link_new(conn, s, &sa, 0)) { - entry->main.timer = uasync_set_timeout(entry->mgr->instance->ua, CONN_MGR_CONNECT_DIRECT_TIMEOUT_MS * 10, entry, cm_direct_timeout_cb, "conn_mgr_direct"); + entry->main.timer = uasync_set_timeout(entry->mgr->instance->ua, (int)(entry->mgr->direct_timeout_ms * 10), entry, cm_direct_timeout_cb, "conn_mgr_direct"); return; } } @@ -469,7 +540,13 @@ static void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) { static void cm_direct_timeout_cb(void* arg) { struct CONN_MGR_ENTRY* entry = (struct CONN_MGR_ENTRY*)arg; entry->main.timer = NULL; - struct TOPO_GROUP* group = topo_groups_get_default(entry->mgr->instance->topo_groups); + if (entry->db_loaded) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: db_node 0x%016llx direct timeout", (unsigned long long)entry->node_id); + cm_cleanup_db_node(entry); + cm_deliver_result(entry, CONN_MGR_ERR_UNREACHABLE); + return; + } + struct TOPO_GROUP* group = entry->mgr->group; struct TOPO_NODEQ* target = topo_node_find_by_id(group, entry->node_id); int has_our = cm_has_direct_ip(group->local_node); int has_target = target ? cm_has_direct_ip(target) : 0; @@ -518,7 +595,7 @@ static void cm_ping_cb(int success, uint16_t rtt, void* arg, uint64_t nonce, ctx->attempt++; if (ctx->attempt < CONN_MGR_LOCAL_SCAN_ATTEMPTS && ctx->phase == 0) { etcp_send_ping_to_socket(entry->mgr->instance, ctx->sock, - topo_node_find_by_id(topo_groups_get_default(entry->mgr->instance->topo_groups), entry->node_id)->node->public_key, + topo_node_find_by_id(entry->mgr->group, entry->node_id)->node->public_key, &ctx->addr, CONN_MGR_LOCAL_SCAN_TIMEOUT_MS, cm_ping_cb, ctx, NULL, 0, 0); return; } @@ -529,7 +606,7 @@ static void cm_ping_cb(int success, uint16_t rtt, void* arg, uint64_t nonce, } static void cm_start_phase_reverse(struct CONN_MGR_ENTRY* entry) { - struct TOPO_GROUP* group = topo_groups_get_default(entry->mgr->instance->topo_groups); + struct TOPO_GROUP* group = entry->mgr->group; struct TOPO_NODEQ* local = group->local_node; if (!local || !local->node) { cm_deliver_result(entry, CONN_MGR_ERR_INTERNAL); return; } uint32_t req_id = ++entry->mgr->next_request_id; @@ -567,7 +644,7 @@ static void cm_start_phase_reverse(struct CONN_MGR_ENTRY* entry) { if (!qe) { u_free(pkt); cm_deliver_result(entry, CONN_MGR_ERR_INTERNAL); return; } qe->dgram = pkt; qe->len = (uint16_t)pkt_size; - etcp_route_send(entry->mgr->instance, TOPO_GROUP_UTUN, entry->node_id, qe, 1); + etcp_route_send(entry->mgr->instance, entry->mgr->group->group_id, entry->node_id, qe, 1); struct cm_reverse_pending* rp = u_calloc(1, sizeof(struct cm_reverse_pending)); if (rp) { rp->request_id = req_id; rp->entry = entry; rp->next = entry->mgr->reverse_pending; @@ -607,6 +684,7 @@ static int cm_start_phase_indirect(struct CONN_MGR_ENTRY* entry) { if (mgr->best_candidate_count == 0) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: phase 3 (indirect) no candidates for 0x%016llx (direct+reverse failed)", (unsigned long long)entry->node_id); + cm_deliver_result(entry, CONN_MGR_ERR_UNREACHABLE); return CONN_MGR_ERR_UNREACHABLE; } uint32_t req_id = ++mgr->next_request_id; @@ -625,7 +703,7 @@ static int cm_start_phase_indirect(struct CONN_MGR_ENTRY* entry) { qe->dgram = u_malloc(sizeof(req)); memcpy(qe->dgram, &req, sizeof(req)); qe->len = (uint16_t)sizeof(req); - etcp_route_send(entry->mgr->instance, TOPO_GROUP_UTUN, entry->node_id, qe, 1); + etcp_route_send(entry->mgr->instance, entry->mgr->group->group_id, entry->node_id, qe, 1); entry->main.timer = uasync_set_timeout(mgr->instance->ua, CONN_MGR_INTERM_EXCHANGE_TIMEOUT_MS * 10, entry, cm_exchange_timeout_cb, "conn_mgr_interm_exch"); { @@ -663,7 +741,7 @@ static void cm_compute_intermediaries(struct CONN_MGR_ENTRY* entry, struct CONN_ } for (uint8_t i = 0; i < resp->my_count && count < 8; i++) { uint64_t id = resp->my_candidates[i].node_id; - struct TOPO_NODEQ* nq = topo_node_find_by_id(topo_groups_get_default(entry->mgr->instance->topo_groups), id); + struct TOPO_NODEQ* nq = topo_node_find_by_id(entry->mgr->group, id); uint16_t our_rtt = nq ? cm_get_node_min_rtt(nq) : 0; if (our_rtt == 0xFFFF) our_rtt = 0; all[count].node_id = id; @@ -689,11 +767,11 @@ static void cm_compute_intermediaries(struct CONN_MGR_ENTRY* entry, struct CONN_ for (uint8_t i = 0; i < sel; i++) { pkt.selected[i].node_id = all[i].node_id; pkt.selected[i].rtt = 0; - struct TOPO_NODEQ* nq = topo_node_find_by_id(topo_groups_get_default(entry->mgr->instance->topo_groups), all[i].node_id); + struct TOPO_NODEQ* nq = topo_node_find_by_id(entry->mgr->group, all[i].node_id); if (nq) pkt.selected[i].rtt = cm_get_node_min_rtt(nq); } struct ll_entry* qe = queue_entry_new(0); - if (qe) { qe->dgram = u_malloc(sizeof(pkt)); memcpy(qe->dgram, &pkt, sizeof(pkt)); qe->len = sizeof(pkt); etcp_route_send(entry->mgr->instance, TOPO_GROUP_UTUN, entry->node_id, qe, 1); } + if (qe) { qe->dgram = u_malloc(sizeof(pkt)); memcpy(qe->dgram, &pkt, sizeof(pkt)); qe->len = sizeof(pkt); etcp_route_send(entry->mgr->instance, entry->mgr->group->group_id, entry->node_id, qe, 1); } entry->conn_type = CONN_TYPE_INDIRECT; entry->state = CONN_MGR_STATE_CONNECTED; @@ -710,7 +788,7 @@ static void cm_exchange_probe_retry_cb(void* arg) { uint8_t need_probe = 0; uint64_t now_tb = get_time_tb(); for (uint8_t i = 0; i < ep->cached_resp.my_count && i < 4; i++) { - struct TOPO_NODEQ* nq = topo_node_find_by_id(topo_groups_get_default(mgr->instance->topo_groups), ep->cached_resp.my_candidates[i].node_id); + struct TOPO_NODEQ* nq = topo_node_find_by_id(mgr->group, ep->cached_resp.my_candidates[i].node_id); if (!nq || !cm_is_rtt_fresh(nq, now_tb)) { need_probe++; break; } } if (need_probe == 0) { @@ -740,7 +818,7 @@ static void cm_handle_interm_exchange_resp(struct CONN_MGR* mgr, const uint8_t* uint8_t need_probe = 0; uint64_t now_tb = get_time_tb(); for (uint8_t i = 0; i < ep->cached_resp.my_count && i < 4; i++) { - struct TOPO_NODEQ* nq = topo_node_find_by_id(topo_groups_get_default(mgr->instance->topo_groups), ep->cached_resp.my_candidates[i].node_id); + struct TOPO_NODEQ* nq = topo_node_find_by_id(mgr->group, ep->cached_resp.my_candidates[i].node_id); if (!nq || !cm_is_rtt_fresh(nq, now_tb)) { need_probe++; route_connectivity_probe_node(mgr->instance, nq); } } if (need_probe == 0) { ep->probes_done = 1; cm_compute_intermediaries(ep->entry, &ep->cached_resp); @@ -829,7 +907,7 @@ static void cm_handle_direct_req(struct ETCP_CONN* conn, const uint8_t* data, si memset(&sa, 0, sizeof(sa)); memcpy(&sa, &sin, sizeof(sin)); struct ETCP_CONN* new_conn = etcp_connection_create(conn->instance, NULL); if (new_conn) { - struct TOPO_NODEQ* nq = topo_node_find_by_id(topo_groups_get_default(conn->instance->topo_groups), src_node_id); + struct TOPO_NODEQ* nq = topo_node_find_by_id(mgr->group, src_node_id); if (nq) { sc_init_ctx(&new_conn->crypto_ctx, &conn->instance->my_keys); sc_set_peer_public_key(&new_conn->crypto_ctx, nq->node->public_key, 0); } struct cm_reverse_pending* rp = u_calloc(1, sizeof(struct cm_reverse_pending)); @@ -851,7 +929,7 @@ static void cm_handle_interm_exchange_req(struct ETCP_CONN* conn, uint64_t now_tb = get_time_tb(); for (uint8_t i = 0; i < req->candidate_count && i < 4; i++) { uint64_t cand_id = req->candidates[i].node_id; - struct TOPO_NODEQ* nq = topo_node_find_by_id(topo_groups_get_default(mgr->instance->topo_groups), cand_id); + struct TOPO_NODEQ* nq = topo_node_find_by_id(mgr->group, cand_id); if (nq && cm_is_rtt_fresh(nq, now_tb)) continue; route_connectivity_probe_node(mgr->instance, nq); } @@ -866,7 +944,7 @@ static void cm_handle_interm_exchange_req(struct ETCP_CONN* conn, resp.your_count = req->candidate_count > 4 ? 4 : req->candidate_count; for (uint8_t i = 0; i < resp.your_count; i++) { resp.your_candidates[i].node_id = req->candidates[i].node_id; - struct TOPO_NODEQ* nq2 = topo_node_find_by_id(topo_groups_get_default(mgr->instance->topo_groups), req->candidates[i].node_id); + struct TOPO_NODEQ* nq2 = topo_node_find_by_id(mgr->group, req->candidates[i].node_id); uint16_t rtt = nq2 ? cm_get_node_min_rtt(nq2) : 0; resp.your_candidates[i].rtt = rtt; } @@ -877,7 +955,7 @@ static void cm_handle_interm_exchange_req(struct ETCP_CONN* conn, qe->dgram = u_malloc(resp_size); memcpy(qe->dgram, &resp, resp_size); qe->len = (uint16_t)resp_size; - etcp_route_send(mgr->instance, TOPO_GROUP_UTUN, conn->peer_node_id, qe, 1); + etcp_route_send(mgr->instance, mgr->group->group_id, conn->peer_node_id, qe, 1); } DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: sent EXCHANGE_RESP to 0x%016llx my=%u your=%u", (unsigned long long)conn->peer_node_id, resp.my_count, resp.your_count); @@ -898,7 +976,7 @@ static void cm_bg_ping_noop_cb(int success, uint16_t rtt, void* arg, uint64_t no static void cm_bg_ping_timer_cb(void* arg) { struct CONN_MGR* mgr = (struct CONN_MGR*)arg; - struct TOPO_GROUP* group = topo_groups_get_default(mgr->instance->topo_groups); + struct TOPO_GROUP* group = mgr->group; if (!group || !group->nodes) { mgr->bg_ping_timer = uasync_set_timeout(mgr->instance->ua, CONN_MGR_BG_PING_INTERVAL_TB, mgr, cm_bg_ping_timer_cb, "conn_mgr_bg_ping"); return; @@ -958,7 +1036,7 @@ static void cm_candidate_ping_timer_cb(void* arg) { uint64_t now_tb = get_time_tb(); for (uint8_t i = 0; i < mgr->best_candidate_count; i++) { uint64_t cand_id = mgr->best_candidates[i].node_id; - struct TOPO_NODEQ* nq = topo_node_find_by_id(topo_groups_get_default(mgr->instance->topo_groups), cand_id); + struct TOPO_NODEQ* nq = topo_node_find_by_id(mgr->group, cand_id); if (!nq) { memmove(&mgr->best_candidates[i], &mgr->best_candidates[i+1], (mgr->best_candidate_count - i - 1) * sizeof(mgr->best_candidates[0])); mgr->best_candidate_count--; i--; continue; @@ -969,7 +1047,7 @@ static void cm_candidate_ping_timer_cb(void* arg) { memmove(&mgr->best_candidates[i], &mgr->best_candidates[i+1], (mgr->best_candidate_count - i - 1) * sizeof(mgr->best_candidates[0])); mgr->best_candidate_count--; i--; continue; } - struct ETCP_CONN* dconn = topo_group_find_conn_for_node(topo_groups_get_default(mgr->instance->topo_groups), cand_id); + struct ETCP_CONN* dconn = topo_group_find_conn_for_node(mgr->group, cand_id); int has_active = 0; if (dconn && dconn->links) { struct ETCP_LINK* l = dconn->links; diff --git a/src/routing_layer/conn_mgr.h b/src/routing_layer/conn_mgr.h index 63231bd5..0c033d1c 100644 --- a/src/routing_layer/conn_mgr.h +++ b/src/routing_layer/conn_mgr.h @@ -276,6 +276,7 @@ struct CONN_MGR_ENTRY { uint8_t local_scan_state; ///< Состояние локального сканирования (enum CM_TRY_STATE) uint8_t main_connect_state; ///< Состояние основного подключения (enum CM_TRY_STATE) + uint8_t db_loaded; ///< 1 = узел загружен из БД, нет BGP-роутинга, только DIRECT struct { uint8_t phase; ///< Текущая фаза: 1=DIRECT, 2=REVERSE, 3=INDIRECT void* timer; ///< Хендл таймера текущей фазы @@ -296,6 +297,7 @@ struct CONN_MGR_ENTRY { */ struct CONN_MGR { struct UTUN_INSTANCE* instance; ///< Родительский экземпляр utun + struct TOPO_GROUP* group; ///< Группа, в которой работает этот менеджер struct CONN_MGR_ENTRY* entries; ///< Динамический массив записей о подключениях size_t entry_count; ///< Текущее количество записей @@ -311,6 +313,7 @@ struct CONN_MGR { uint8_t best_candidate_count; ///< Количество актуальных кандидатов (≤CONN_MGR_MAX_CANDIDATES) uint32_t next_request_id; ///< Счётчик request_id для INTERM_EXCHANGE + uint32_t direct_timeout_ms; ///< Таймаут фазы DIRECT (мс), по умолчанию CONN_MGR_CONNECT_DIRECT_TIMEOUT_MS uint8_t initialized; ///< Флаг инициализации struct cm_reverse_pending* reverse_pending; ///< Связный список ожидающих REVERSE-подключений @@ -320,14 +323,14 @@ struct CONN_MGR { /* ==================== Жизненный цикл ==================== */ /** - * @brief Инициализирует менеджер соединений. - * @param instance родительский экземпляр utun + * @brief Инициализирует менеджер соединений для заданной группы. + * @param group группа топологии (conn_mgr будет работать в её контексте) * @return указатель на CONN_MGR или NULL при ошибке * * Регистрирует обработчик в etcp_router для ETCP_RT_ID_CONN_MGR, * запускает фоновые таймеры (bg_ping, candidate_ping). */ -struct CONN_MGR* conn_mgr_init(struct UTUN_INSTANCE* instance); +struct CONN_MGR* conn_mgr_init(struct TOPO_GROUP* group); /** * @brief Уничтожает менеджер соединений. @@ -382,6 +385,16 @@ int conn_mgr_disconnect_node(struct CONN_MGR* mgr, uint64_t node_id); */ int conn_mgr_set_idle_timeout(struct CONN_MGR* mgr, uint64_t node_id, uint32_t timeout_ms); +/** + * @brief Устанавливает таймаут фазы DIRECT для новых подключений. + * @param mgr менеджер + * @param timeout_ms таймаут (мс), по умолчанию CONN_MGR_CONNECT_DIRECT_TIMEOUT_MS (5000) + * + * Устанавливается глобально для всех новых вызовов conn_mgr_connect_node(). + * Действует на оба пути: обычный (BGP) и загруженный из БД (db_loaded). + */ +void conn_mgr_set_direct_timeout_ms(struct CONN_MGR* mgr, uint32_t timeout_ms); + /** * @brief Получает текущий статус подключения к ноде. * @param mgr менеджер diff --git a/src/routing_layer/nat_detection.c b/src/routing_layer/nat_detection.c index 1e8b1915..37bbbeee 100644 --- a/src/routing_layer/nat_detection.c +++ b/src/routing_layer/nat_detection.c @@ -193,7 +193,7 @@ static void nat_detection_handle_nat_info(struct NAT_DETECTION* nd, struct TOPO_ while (se) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data; if (item && item->conn) - topo_group_send_nodeinfo(group, group->local_node, item->conn); + topo_group_send_nodeinfo(group, group->local_node, item->conn, 0); se = se->next; } } diff --git a/src/routing_layer/topo_group.c b/src/routing_layer/topo_group.c index e957f82c..c99be2bf 100644 --- a/src/routing_layer/topo_group.c +++ b/src/routing_layer/topo_group.c @@ -22,6 +22,7 @@ #include "nat_detection.h" #include "route_connectivity.h" #include "control_server.h" +#include "conn_mgr.h" #define TOPO_GROUP_UTUN 0x8000000000000000ULL @@ -225,6 +226,8 @@ static struct TOPO_GROUP* topo_group_create(struct UTUN_INSTANCE* instance, uint group->local_node = NULL; topo_group_update_my_nodeinfo(instance, group); + group->conn_mgr = conn_mgr_init(group); + DEBUG_INFO(DEBUG_CATEGORY_BGP, "TOPO group created: group_id=%016llx", (unsigned long long)group_id); return group; } @@ -240,6 +243,7 @@ static void topo_group_destroy(struct TOPO_GROUP* group) { if (group->local_node) { topo_node_free_lists(group, group->local_node); u_free(group->local_node); } if (group->nodes) queue_free(group->nodes); + group->conn_mgr = NULL; /* destroyed externally in utun_instance_destroy before etcp_router_destroy */ } struct TOPO_GROUPS* topo_groups_init(struct UTUN_INSTANCE* instance) { @@ -297,6 +301,8 @@ struct TOPO_GROUPS* topo_groups_init(struct UTUN_INSTANCE* instance) { } queue_data_put_with_index(g->group_list, &default_group->ll); + instance->conn_mgr = default_group->conn_mgr; /* recv handler shortcut */ + etcp_bind(instance, ETCP_ID_TOPO_ENTRY, topo_group_receive_cbk); etcp_add_conn_status_cbk(instance, topo_group_conn_status, g); @@ -460,13 +466,13 @@ struct ETCP_CONN* topo_group_find_conn_for_node(struct TOPO_GROUP* group, uint64 return best_live ? best_live->conn : (best_any ? best_any->conn : NULL); } -int topo_group_add_path(struct TOPO_NODEQ* nq, struct ETCP_CONN* conn, uint64_t* hop_list, uint8_t hop_count) { +int topo_group_add_path(struct TOPO_NODEQ* nq, struct ETCP_CONN* conn, uint64_t* hop_list, uint8_t hop_count, uint16_t cumulative_rtt) { if (!nq || !conn || hop_count > MAX_HOPS || !hop_list) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "add_path: invalid args"); return -1; } if (!nq->paths) { nq->paths = queue_new(conn->instance->ua, 0, 0, 0, "node_paths"); if (!nq->paths) return -1; } size_t path_size = sizeof(struct TOPO_NODEPATH) - sizeof(struct ll_entry) + hop_count * 8; struct ll_entry* pe = queue_entry_new(path_size); if (!pe) return -1; struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)pe; - path->conn = conn; path->hop_count = hop_count; + path->conn = conn; path->hop_count = hop_count; path->cumulative_rtt = cumulative_rtt; memcpy((uint64_t*)((uint8_t*)path + sizeof(struct TOPO_NODEPATH)), hop_list, hop_count * 8); queue_data_put(nq->paths, pe); return 0; @@ -551,7 +557,7 @@ int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from memcpy(hop_list, hop_src, ni->hop_count * 8); hop_list[ni->hop_count] = from->peer_node_id; topo_group_remove_path(nodeinfo1, from); - topo_group_add_path(nodeinfo1, from, hop_list, new_hops); + topo_group_add_path(nodeinfo1, from, hop_list, new_hops, nodeinfo1->cumulative_rtt); } return 0; } @@ -573,7 +579,8 @@ int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from struct TOPO_NODESUBNETS* new_subnets = NULL; struct TOPOMSG_TRANZIT* new_tranzit = NULL; uint8_t new_tranzit_count = 0; uint64_t* new_hop_list = NULL; uint8_t new_hop_count = 0; - if (topo_node_deserialize(group, ser_data, ser_len, &new_ni, &new_subnets, &new_tranzit, &new_tranzit_count, &new_hop_list, &new_hop_count) != 0) { + uint16_t incoming_cumulative_rtt = 0; + if (topo_node_deserialize(group, ser_data, ser_len, &new_ni, &new_subnets, &new_tranzit, &new_tranzit_count, &new_hop_list, &new_hop_count, &incoming_cumulative_rtt) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "NODEINFO deserialize failed from %s", from->log_name); if (nodeinfo1) { queue_remove_data(group->nodes, &nodeinfo1->ll); queue_free(paths); queue_entry_free(&nodeinfo1->ll); } return -1; @@ -583,6 +590,7 @@ int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from nodeinfo1->node = topo_node_registry_acquire(group->instance->topo_groups, new_ni); nodeinfo1->hash_node_id = nodeinfo1->node ? nodeinfo1->node->node_id : 0; nodeinfo1->subnets = new_subnets; + nodeinfo1->cumulative_rtt = incoming_cumulative_rtt; u_free(nodeinfo1->tranzit_data); nodeinfo1->tranzit_data = new_tranzit; nodeinfo1->tranzit_count = new_tranzit_count; u_free(nodeinfo1->hop_list); nodeinfo1->hop_list = new_hop_list; @@ -595,6 +603,7 @@ int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from nodeinfo1->node = topo_node_registry_acquire(group->instance->topo_groups, new_ni); nodeinfo1->hash_node_id = nodeinfo1->node ? nodeinfo1->node->node_id : 0; nodeinfo1->subnets = new_subnets; + nodeinfo1->cumulative_rtt = incoming_cumulative_rtt; nodeinfo1->tranzit_data = new_tranzit; nodeinfo1->tranzit_count = new_tranzit_count; nodeinfo1->hop_list = new_hop_list; nodeinfo1->hop_count = new_hop_count; nodeinfo1->connectivity.probe_status = PROBE_STATUS_NONE; @@ -617,7 +626,7 @@ int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from memcpy(nodeinfo1->hop_list, hop_list, nodeinfo1->hop_count * 8); topo_group_remove_path_by_hop(nodeinfo1, from->peer_node_id); - topo_group_add_path(nodeinfo1, from, hop_list, nodeinfo1->hop_count); + topo_group_add_path(nodeinfo1, from, hop_list, nodeinfo1->hop_count, incoming_cumulative_rtt); if (group->group_type != TOPO_GROUP_TYPE_CHAT && group->instance->rt) route_insert(group->instance->rt, nodeinfo1); if (node_id != group->instance->node_id) { @@ -645,7 +654,10 @@ int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from uint64_t id = item->conn->peer_node_id; if (id != node_id) { int found = 0; for (int i = 0; i < hop_count; i++) if (hop_list[i] == id) found = 1; - if (found == 0) topo_group_send_nodeinfo(group, nodeinfo1, item->conn); + if (found == 0) { + uint16_t out_rtt = (uint16_t)((uint32_t)item->conn->rtt_last + incoming_cumulative_rtt); + topo_group_send_nodeinfo(group, nodeinfo1, item->conn, out_rtt); + } } } se = se->next; @@ -687,7 +699,7 @@ int topo_group_process_withdraw(struct TOPO_GROUP* group, struct ETCP_CONN* send return 0; } -void topo_group_send_nodeinfo(struct TOPO_GROUP* group, struct TOPO_NODEQ* node, struct ETCP_CONN* conn) { +void topo_group_send_nodeinfo(struct TOPO_GROUP* group, struct TOPO_NODEQ* node, struct ETCP_CONN* conn, uint16_t cumulative_rtt) { if (!group || !node || !conn) return; size_t max_sz = sizeof(struct TOPOMSG_NODEINFO_PKT) + 4096; @@ -696,7 +708,7 @@ void topo_group_send_nodeinfo(struct TOPO_GROUP* group, struct TOPO_NODEQ* node, p[0] = ETCP_ID_TOPO_ENTRY; p[1] = TOPO_SUBCMD_NODEINFO; - int ser_len = topo_node_serialize(group, node, p + 2, max_sz - 2); + int ser_len = topo_node_serialize(group, node, p + 2, max_sz - 2, cumulative_rtt); if (ser_len < 0) { u_free(p); return; } struct ll_entry* e = queue_entry_new(0); @@ -739,14 +751,14 @@ static void topo_group_send_full_table(struct TOPO_GROUP* group, struct ETCP_CON struct TOPO_NODEQ* nq = (struct TOPO_NODEQ*)e; if (nq->hop_count == 0) { e = e->next; continue; } if (!nq->paths) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "node has no paths"); e = e->next; continue; } - if (topo_group_should_send_to(nq, target)) topo_group_send_nodeinfo(group, nq, conn); + if (topo_group_should_send_to(nq, target)) topo_group_send_nodeinfo(group, nq, conn, (uint16_t)((uint32_t)conn->rtt_last + nq->cumulative_rtt)); e = e->next; } } static void topo_group_handle_request_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { if (!group || !conn) return; - topo_group_send_nodeinfo(group, group->local_node, conn); + topo_group_send_nodeinfo(group, group->local_node, conn, 0); topo_group_send_full_table(group, conn); topo_group_add_to_senders(group, conn); topo_group_send_table_complete(group, conn); diff --git a/src/routing_layer/topo_group.h b/src/routing_layer/topo_group.h index 3f5d47d5..25068ab7 100644 --- a/src/routing_layer/topo_group.h +++ b/src/routing_layer/topo_group.h @@ -41,6 +41,7 @@ extern "C" { struct UTUN_INSTANCE; struct NAT_DETECTION; +struct CONN_MGR; /** Callback when a node's info (pubkeys, addresses) is persisted in the DB */ typedef void (*topo_node_updated_fn)(struct UTUN_INSTANCE* inst, uint64_t node_id, @@ -128,6 +129,7 @@ struct TOPO_GROUP { struct TOPO_NODEQ* local_node; uint8_t ed25519_public_key[SC_PUBKEY_SIZE]; char channel_id[64]; // channel_id для групп типа CHAT + struct CONN_MGR* conn_mgr; // менеджер соединений для этой группы }; /** @@ -218,7 +220,7 @@ int topo_group_process_withdraw(struct TOPO_GROUP* group, struct ETCP_CONN* send /** * @brief Отправляет NODEINFO пакет одному conn (всегда local_node). */ -void topo_group_send_nodeinfo(struct TOPO_GROUP* group, struct TOPO_NODEQ* node, struct ETCP_CONN* conn); +void topo_group_send_nodeinfo(struct TOPO_GROUP* group, struct TOPO_NODEQ* node, struct ETCP_CONN* conn, uint16_t cumulative_rtt); /** * @brief Отправляет WITHDRAW для node_id (вызывает broadcast_withdraw). @@ -241,7 +243,7 @@ struct ETCP_CONN* topo_group_find_conn_for_node(struct TOPO_GROUP* group, uint64 * * @return 0 при успехе */ -int topo_group_add_path(struct TOPO_NODEQ* nq, struct ETCP_CONN* conn, uint64_t* hop_list, uint8_t hop_count); +int topo_group_add_path(struct TOPO_NODEQ* nq, struct ETCP_CONN* conn, uint64_t* hop_list, uint8_t hop_count, uint16_t cumulative_rtt); /** * @brief Удаляет conn из paths узла. diff --git a/src/routing_layer/topo_node.c b/src/routing_layer/topo_node.c index 03edb123..6accf21b 100644 --- a/src/routing_layer/topo_node.c +++ b/src/routing_layer/topo_node.c @@ -137,7 +137,7 @@ int topo_node_dyn_size(const struct TOPOMSG_NODE* msg) { + msg->hop_count * 8; } -int topo_node_serialize(struct TOPO_GROUP* group, struct TOPO_NODEQ* nq, uint8_t* out, size_t out_max) { +int topo_node_serialize(struct TOPO_GROUP* group, struct TOPO_NODEQ* nq, uint8_t* out, size_t out_max, uint16_t cumulative_rtt) { if (!group || !nq || !nq->node || !out) return -1; struct TOPO_NODE* ni = nq->node; @@ -158,6 +158,7 @@ int topo_node_serialize(struct TOPO_GROUP* group, struct TOPO_NODEQ* nq, uint8_t msg.local_v6_subnets = nq->subnets ? topo_list_count((struct _topo_head*)nq->subnets->v6_subnets) : 0; msg.tranzit_nodes = nq->tranzit_count; msg.hop_count = nq->hop_count; + msg.cumulative_rtt = cumulative_rtt; size_t total = TOPOMSG_NODE_HDR_SIZE + (size_t)topo_node_dyn_size(&msg); if (total > out_max) return -1; @@ -210,7 +211,8 @@ int topo_node_serialize(struct TOPO_GROUP* group, struct TOPO_NODEQ* nq, uint8_t int topo_node_deserialize(struct TOPO_GROUP* group, const uint8_t* data, size_t len, struct TOPO_NODE** out_ni, struct TOPO_NODESUBNETS** out_subnets, struct TOPOMSG_TRANZIT** out_tranzit, uint8_t* out_tranzit_count, - uint64_t** out_hop_list, uint8_t* out_hop_count) { + uint64_t** out_hop_list, uint8_t* out_hop_count, + uint16_t* out_cumulative_rtt) { if (!group || !data || len < TOPOMSG_NODE_HDR_SIZE || !out_ni) return -1; const struct TOPOMSG_NODE* msg = (const struct TOPOMSG_NODE*)data; @@ -305,6 +307,7 @@ int topo_node_deserialize(struct TOPO_GROUP* group, const uint8_t* data, size_t *out_hop_count = msg->hop_count; } + if (out_cumulative_rtt) *out_cumulative_rtt = msg->cumulative_rtt; *out_ni = ni; return 0; } diff --git a/src/routing_layer/topo_node.h b/src/routing_layer/topo_node.h index f0acec43..35e2a403 100644 --- a/src/routing_layer/topo_node.h +++ b/src/routing_layer/topo_node.h @@ -174,6 +174,7 @@ struct TOPOMSG_NODE { uint8_t local_v6_subnets; uint8_t tranzit_nodes; uint8_t hop_count; + uint16_t cumulative_rtt; // сумма RTT upstream (0.1ms), без последнего линка } __attribute__((packed)); #define TOPOMSG_NODE_HDR_SIZE sizeof(struct TOPOMSG_NODE) @@ -224,6 +225,7 @@ struct TOPO_NODEPATH { struct ll_entry ll; struct ETCP_CONN* conn; uint8_t hop_count; + uint16_t cumulative_rtt; // сумма RTT upstream (0.1ms) для этого пути }; struct TOPO_NODEQ { @@ -235,6 +237,7 @@ struct TOPO_NODEQ { uint64_t* hop_list; // отдельный malloc uint8_t tranzit_count; uint8_t hop_count; + uint16_t cumulative_rtt; // база из последнего NODEINFO для пересылки struct ll_queue* paths; uint8_t dirty; uint8_t last_ver; @@ -277,9 +280,10 @@ static inline const struct TOPO_ADDR6* topo_v6_addrs(const struct TOPO_NODE* static inline const struct TOPO_SUBNET4* topo_v4_subnets(const struct TOPO_NODESUBNETS* r) { return r ? r->v4_subnets : NULL; } static inline const struct TOPO_SUBNET6* topo_v6_subnets(const struct TOPO_NODESUBNETS* r) { return r ? r->v6_subnets : NULL; } -int topo_node_serialize(struct TOPO_GROUP* group, struct TOPO_NODEQ* nq, uint8_t* out, size_t out_max); +int topo_node_serialize(struct TOPO_GROUP* group, struct TOPO_NODEQ* nq, uint8_t* out, size_t out_max, uint16_t cumulative_rtt); int topo_node_deserialize(struct TOPO_GROUP* group, const uint8_t* data, size_t len, struct TOPO_NODE** out_ni, struct TOPO_NODESUBNETS** out_subnets, - struct TOPOMSG_TRANZIT** out_tranzit, uint8_t* out_tranzit_count, uint64_t** out_hop_list, uint8_t* out_hop_count); + struct TOPOMSG_TRANZIT** out_tranzit, uint8_t* out_tranzit_count, uint64_t** out_hop_list, uint8_t* out_hop_count, + uint16_t* out_cumulative_rtt); int topo_group_update_my_nodeinfo(struct UTUN_INSTANCE* instance, struct TOPO_GROUP* group); diff --git a/src/routing_layer/topo_node_sqlite.c b/src/routing_layer/topo_node_sqlite.c index f430dbdd..7d3729c9 100644 --- a/src/routing_layer/topo_node_sqlite.c +++ b/src/routing_layer/topo_node_sqlite.c @@ -1,8 +1,10 @@ #include "topo_node_sqlite.h" #include "topo_node.h" +#include "topo_group.h" #include "etcp_connections.h" #include "../lib/debug_config.h" #include "../lib/mem.h" +#include "../lib/memory_pool.h" #include #include @@ -655,3 +657,70 @@ void topo_node_sqlite_update_rtt(sqlite3* db, uint64_t node_id, uint16_t rtt) { DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "sqlite_update_rtt: node=%016llx rtt=%u", (unsigned long long)node_id, (unsigned)rtt); } + +struct TOPO_NODE* topo_node_sqlite_node_load(sqlite3* db, struct TOPO_GROUPS* groups, uint64_t node_id) { + if (!db || !groups) return NULL; + + uint8_t pubkey[32] = {0}, ed_pubkey[32] = {0}; + char* name = NULL; + int found = 0; + { + sqlite3_stmt* s = NULL; + if (sqlite3_prepare_v2(db, "SELECT name,x25519_pubkey,ed25519_pubkey FROM nodes WHERE node_id=?", + -1, &s, NULL) == SQLITE_OK) { + sqlite3_bind_int64(s, 1, (sqlite3_int64)node_id); + if (sqlite3_step(s) == SQLITE_ROW) { + const char* n = (const char*)sqlite3_column_text(s, 0); + if (n) name = u_strdup(n); + const void* pk = sqlite3_column_blob(s, 1); + if (pk && sqlite3_column_bytes(s, 1) >= 32) memcpy(pubkey, pk, 32); + const void* ep = sqlite3_column_blob(s, 2); + if (ep && sqlite3_column_bytes(s, 2) >= 32) memcpy(ed_pubkey, ep, 32); + found = 1; + } + sqlite3_finalize(s); + } + } + if (!found) { u_free(name); return NULL; } + + struct TOPO_ADDR4* v4_head = NULL; struct TOPO_ADDR4* v4_tail = NULL; + uint8_t sock_id_counter = 0; + { + sqlite3_stmt* s = NULL; + if (sqlite3_prepare_v2(db, + "SELECT address,port,addr_type,socket_id FROM node_addresses" + " WHERE node_id=? AND family=4 AND protocol=1 ORDER BY addr_type", + -1, &s, NULL) == SQLITE_OK) { + sqlite3_bind_int64(s, 1, (sqlite3_int64)node_id); + while (sqlite3_step(s) == SQLITE_ROW) { + const void* addr = sqlite3_column_blob(s, 0); + if (!addr || sqlite3_column_bytes(s, 0) < 4) continue; + int port = sqlite3_column_int(s, 1); + int at = sqlite3_column_int(s, 2); + int sock_id_raw = sqlite3_column_int(s, 3); + uint8_t sock_id = (uint8_t)(sock_id_raw ? sock_id_raw : ++sock_id_counter); + uint8_t type = (at == ADDR_TYPE_NETIF) ? TOPO_ADDR_INTERFACE : TOPO_ADDR_NAT; + struct TOPO_ADDR4* a4 = memory_pool_alloc(groups->v4_addr_pool); + if (!a4) continue; + memset(a4, 0, sizeof(*a4)); + memcpy(a4->addr, addr, 4); a4->port = (uint16_t)port; + a4->type = type; a4->socket_id = sock_id; a4->protocol = TOPO_PROTO_UDP; + if (!v4_head) v4_head = v4_tail = a4; + else { v4_tail->next = a4; v4_tail = a4; } + } + sqlite3_finalize(s); + } + } + if (!v4_head) { u_free(name); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "node_load: node 0x%016llx has no IPv4/UDP addresses", (unsigned long long)node_id); return NULL; } + + struct TOPO_NODE* ni = u_calloc(1, sizeof(struct TOPO_NODE)); + if (!ni) { u_free(name); return NULL; } + ni->node_id = node_id; ni->ver = 1; ni->ref_count = 0; + memcpy(ni->public_key, pubkey, 32); memcpy(ni->ed25519_public_key, ed_pubkey, 32); + ni->node_name = name; + ni->v4_addrs = v4_head; + + DEBUG_INFO(DEBUG_CATEGORY_BGP, "node_load: node 0x%016llx loaded from DB name=%s v4_addrs=%d", + (unsigned long long)node_id, name ? name : "", topo_list_count((struct _topo_head*)v4_head)); + return ni; +} diff --git a/src/routing_layer/topo_node_sqlite.h b/src/routing_layer/topo_node_sqlite.h index 3f95126f..cda6d134 100644 --- a/src/routing_layer/topo_node_sqlite.h +++ b/src/routing_layer/topo_node_sqlite.h @@ -7,6 +7,7 @@ #include struct TOPO_NODEQ; +struct TOPO_GROUPS; #define PEERS_JOIN_SIG_SIZE 64 #define PEERS_JOIN_TS_SIZE 8 @@ -56,4 +57,14 @@ int topo_node_sqlite_nodeinfo_updated(sqlite3* db, uint64_t node_id); /** Обновляет RTT узла: node_addresses.rtt и все peers_*.node_RTT */ void topo_node_sqlite_update_rtt(sqlite3* db, uint64_t node_id, uint16_t rtt); +/** + * @brief Загружает TOPO_NODE из SQLite (ref_count=0). + * @param groups нужен для выделения адресов из memory_pool + * @return TOPO_NODE* или NULL если узел не найден/нет адресов + * + * Загружает pubkey, name из nodes, IPv4/UDP-адреса из node_addresses. + * Вызывающий должен передать владение через topo_node_registry_acquire(). + */ +struct TOPO_NODE* topo_node_sqlite_node_load(sqlite3* db, struct TOPO_GROUPS* groups, uint64_t node_id); + #endif diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c index 70d0d0cd..be93fbd5 100644 --- a/src/transport_layer/etcp_connections.c +++ b/src/transport_layer/etcp_connections.c @@ -1323,7 +1323,7 @@ static void send_init_response(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pk struct ll_entry* se = g->senders_list->head; while (se) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data; - if (item && item->conn) topo_group_send_nodeinfo(g, g->local_node, item->conn); + if (item && item->conn) topo_group_send_nodeinfo(g, g->local_node, item->conn, 0); se = se->next; } } diff --git a/src/utun_instance.c b/src/utun_instance.c index 146ad5d4..0b6fb6e5 100644 --- a/src/utun_instance.c +++ b/src/utun_instance.c @@ -205,8 +205,7 @@ static int instance_init_common(struct UTUN_INSTANCE* instance, struct UASYNC* u instance->topo_groups = NULL; } - instance->conn_mgr = conn_mgr_init(instance); - if (!instance->conn_mgr) DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "Failed to initialize Connection Manager (non-fatal)"); + // conn_mgr initialized inside topo_group_create (via topo_groups_init) // Initialize firewall fw_init(&instance->fw); @@ -483,10 +482,11 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) { nat_transport_destroy(instance); } + // Cleanup conn_mgr (must be before etcp_router_destroy — unbinds from router) + if (instance->conn_mgr) { conn_mgr_destroy(instance->conn_mgr); instance->conn_mgr = NULL; } + // Cleanup etcp_router etcp_router_destroy(instance); - - if (instance->conn_mgr) { conn_mgr_destroy(instance->conn_mgr); instance->conn_mgr = NULL; } // Cleanup BGP module if (instance->topo_groups) { diff --git a/tests/test_conn_mgr.c b/tests/test_conn_mgr.c index a6673d2e..abfc38fa 100644 --- a/tests/test_conn_mgr.c +++ b/tests/test_conn_mgr.c @@ -1,14 +1,11 @@ /** * @file test_conn_mgr.c - * @brief Connection Manager — smoke test + * @brief Connection Manager — smoke + DB tests * * Test 1: Direct via existing ETCP conn → CONN_TYPE_DIRECT - * Test 2: Status check → state/type - * - * Pending: - * Test 4: New link UP callback (INIT to new peer via conn_mgr) - * Test 5: Indirect through intermediary (needs full Phase 3 exchange) - * Test 6: Idle timeout + * Test 2: Status check → state/type, correct topo_group + * Test 3: DB-loaded node → successful DIRECT connect, node stays in group + * Test 4: DB-loaded node → unreachable → UNREACHABLE + node removed from group */ #include #include @@ -33,6 +30,7 @@ #define TIMEOUT_TB 300000 #define POLL_MS 5 +#define DB_TEST_TIMEOUT_MS 200 static uint64_t g_nid_a = 0; static uint64_t g_nid_b = 0; @@ -54,18 +52,13 @@ static char* gv(const char* p, const char* k) { char* r = (strcmp(k, "pub") == 0) ? u_strdup(c->global.my_public_key_hex) : u_strdup(c->global.my_private_key_hex); free_config(c); return r; } -static int lks(struct UTUN_INSTANCE* i) { - int n = 0; struct ll_entry* entry = i->connections->head; - while (entry) { struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data; - struct ETCP_CONN* c = ce->conn; - struct ETCP_LINK* l = c->links; while (l) { if (l->initialized && l->link_status) n++; l = l->next; } - entry = entry->next; } - return n; -} static void fail(const char* msg) { fprintf(stderr, "FAIL: %s\n", msg); fflush(stderr); result = 2; } static void test1(void* arg); static void test2(void* arg); static void test3(void* arg); +static void test3_check(void* arg); +static void test4(void* arg); +static void test4_check(void* arg); static void to_cb(void* arg) { (void)arg; fprintf(stderr, "TIMEOUT\n"); result = 2; } static void ccb(int r, uint64_t id, void* arg) { @@ -83,15 +76,17 @@ static void test1(void* arg) { static void test2(void* arg) { (void)arg; if (result) return; if (!cdone) { uasync_set_timeout(ua, 50, NULL, (timeout_callback_t)test2, "t2b"); return; } - if (cresult != CONN_MGR_OK) { fail("connect failed"); return; } + if (cresult != CONN_MGR_OK) { fail("test2: connect failed"); return; } uint8_t st, ty; conn_mgr_get_status(g_a->conn_mgr, g_nid_b, &st, &ty); fprintf(stderr, "Test 2: status state=%d type=%d\n", st, ty); fflush(stderr); - if (st == CONN_MGR_STATE_CONNECTED && ty == CONN_TYPE_DIRECT) fprintf(stderr, " OK: DIRECT\n"); - else { fail("status mismatch"); return; } - fprintf(stderr, "=== ALL DONE ===\n"); fflush(stderr); + if (st != CONN_MGR_STATE_CONNECTED || ty != CONN_TYPE_DIRECT) { fail("test2: status mismatch"); return; } + if (g_a->conn_mgr->group->group_id != TOPO_GROUP_UTUN) { fail("test2: wrong group_id"); return; } + fprintf(stderr, " OK: DIRECT, group_id correct\n"); fflush(stderr); + fprintf(stderr, "=== PHASE 1 DONE ===\n"); fflush(stderr); result = (result == 0) ? 1 : 2; } +/* ---- Phase 1: setup/cleanup ---- */ static void setup(void) { test_mkdtemp(tdir); int base = 47000 + (getpid() % 15000); pa = base; pb = base + 1; @@ -109,22 +104,183 @@ static void setup(void) { } static void cleanup(void) { test_unlink(ca); test_unlink(cb); test_rmdir(tdir); } +/* ---- Phase 2: DB tests ---- */ +static struct UTUN_INSTANCE* g_a2 = NULL, *g_b2 = NULL; +static struct UASYNC* ua2 = NULL; +static uint64_t g_nid_a2 = 0, g_nid_b2 = 0, g_nid_c = 0; +static volatile int cdone2 = 0, cresult2 = 0; +static char tdir2[] = "/tmp/utun_cm_db_XXXXXX"; +static char ca2[256], cb2[256]; +static int pa2 = 0, pb2 = 0; + +static void ccb2(int r, uint64_t id, void* arg) { + (void)arg; fprintf(stderr, "connect_cb2: result=%d node=0x%llx\n", r, (unsigned long long)id); fflush(stderr); + cdone2 = 1; cresult2 = r; +} + +static int hex2bin(const char* hex, uint8_t* out, size_t out_len) { + size_t len = strlen(hex); + if (len > out_len * 2) return -1; + for (size_t i = 0; i < len; i += 2) { + unsigned int b; + if (sscanf(hex + i, "%2x", &b) != 1) return -1; + out[i / 2] = (uint8_t)b; + } + return 0; +} + +static int db_insert_node(sqlite3* db, uint64_t node_id, const uint8_t* x25519_pub, + const char* name, uint8_t ip[4], int port) { + sqlite3_stmt* s = NULL; + uint8_t zero32[32]; memset(zero32, 0, 32); + if (sqlite3_prepare_v2(db, + "INSERT OR REPLACE INTO nodes(node_id,name,x25519_pubkey,ed25519_pubkey,last_seen_at) VALUES(?,?,?,?,unixepoch())", + -1, &s, NULL) != SQLITE_OK) { fprintf(stderr, "prepare nodes: %s\n", sqlite3_errmsg(db)); return -1; } + sqlite3_bind_int64(s, 1, (sqlite3_int64)node_id); + sqlite3_bind_text(s, 2, name, -1, SQLITE_STATIC); + sqlite3_bind_blob(s, 3, x25519_pub, 32, SQLITE_STATIC); + sqlite3_bind_blob(s, 4, zero32, 32, SQLITE_STATIC); + if (sqlite3_step(s) != SQLITE_DONE) { fprintf(stderr, "insert nodes: %s\n", sqlite3_errmsg(db)); sqlite3_finalize(s); return -1; } + sqlite3_finalize(s); + + if (sqlite3_prepare_v2(db, + "INSERT INTO node_addresses(node_id,family,protocol,address,port,addr_type,socket_id) VALUES(?,4,1,?,?,?,0)", + -1, &s, NULL) != SQLITE_OK) { fprintf(stderr, "prepare addrs: %s\n", sqlite3_errmsg(db)); return -1; } + sqlite3_bind_int64(s, 1, (sqlite3_int64)node_id); + sqlite3_bind_blob(s, 2, ip, 4, SQLITE_STATIC); + sqlite3_bind_int(s, 3, port); + sqlite3_bind_int(s, 4, ADDR_TYPE_DIRECT); + if (sqlite3_step(s) != SQLITE_DONE) { fprintf(stderr, "insert addrs: %s\n", sqlite3_errmsg(db)); sqlite3_finalize(s); return -1; } + sqlite3_finalize(s); + return 0; +} + +static void test3(void* arg) { + (void)arg; if (result) return; + fprintf(stderr, "Test 3: DB node (B2) — direct connect, node stays in group\n"); fflush(stderr); + cdone2 = 0; cresult2 = 0; + conn_mgr_connect_node(g_a2->conn_mgr, g_nid_b2, 0, ccb2, NULL); + uasync_set_timeout(ua2, 20, NULL, (timeout_callback_t)test3_check, "t3a"); +} +static void test3_check(void* arg) { + (void)arg; + if (!cdone2) { uasync_set_timeout(ua2, 20, NULL, (timeout_callback_t)test3_check, "t3b"); return; } + if (cresult2 != CONN_MGR_OK) { fail("test3: connect B2 failed"); return; } + + uint8_t st, ty; conn_mgr_get_status(g_a2->conn_mgr, g_nid_b2, &st, &ty); + fprintf(stderr, " state=%d type=%d\n", st, ty); fflush(stderr); + if (st != CONN_MGR_STATE_CONNECTED || ty != CONN_TYPE_DIRECT) { fail("test3: status mismatch"); return; } + + struct TOPO_NODEQ* nq = topo_node_find_by_id(g_a2->conn_mgr->group, g_nid_b2); + if (!nq) { fail("test3: B2 not in group after connect"); return; } + if (g_a2->conn_mgr->group->group_id != TOPO_GROUP_UTUN) { fail("test3: wrong group_id"); return; } + + fprintf(stderr, " OK: DIRECT, node in group, group_id=%016llx\n", (unsigned long long)g_a2->conn_mgr->group->group_id); fflush(stderr); + cdone2 = 0; cresult2 = 0; + uasync_set_timeout(ua2, 50, NULL, (timeout_callback_t)test4, "t4a"); +} + +static void test4(void* arg) { + (void)arg; if (result) return; + fprintf(stderr, "Test 4: DB node (C, unreachable) — timeout → UNREACHABLE, node removed\n"); fflush(stderr); + cdone2 = 0; cresult2 = 0; + conn_mgr_connect_node(g_a2->conn_mgr, g_nid_c, 0, ccb2, NULL); + uasync_set_timeout(ua2, 20, NULL, (timeout_callback_t)test4_check, "t4b"); +} +static void test4_check(void* arg) { + (void)arg; + if (!cdone2) { uasync_set_timeout(ua2, 20, NULL, (timeout_callback_t)test4_check, "t4c"); return; } + if (cresult2 != CONN_MGR_ERR_UNREACHABLE) { fail("test4: expected UNREACHABLE"); return; } + + struct TOPO_NODEQ* nq = topo_node_find_by_id(g_a2->conn_mgr->group, g_nid_c); + if (nq && !(nq->paths && queue_entry_count(nq->paths) > 0)) { fail("test4: C still in group after unreachable"); return; } + + fprintf(stderr, " OK: UNREACHABLE, node cleaned up\n"); fflush(stderr); + fprintf(stderr, "=== ALL DONE ===\n"); fflush(stderr); + result = (result == 0) ? 1 : 2; +} + +static void setup_db_test(void) { + test_mkdtemp(tdir2); + int base = 48000 + (getpid() % 10000); pa2 = base; pb2 = base + 1; + char dbdir_a[320], dbdir_b[320]; + snprintf(dbdir_a, sizeof(dbdir_a), "%s/db_a", tdir2); utun_mkdir(dbdir_a, 0755); + snprintf(dbdir_b, sizeof(dbdir_b), "%s/db_b", tdir2); utun_mkdir(dbdir_b, 0755); + snprintf(ca2, sizeof(ca2), "%s/a2.conf", tdir2); snprintf(cb2, sizeof(cb2), "%s/b2.conf", tdir2); + wf(ca2, "[global]\ntun_ip=10.98.0.1/24\ntun_ifname=tun97\n" + "db_path=%s\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", dbdir_a, pa2); + wf(cb2, "[global]\ntun_ip=10.98.0.2/24\ntun_ifname=tun96\n" + "db_path=%s\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", dbdir_b, pb2); + config_ensure_keys_and_node_id(ca2); config_ensure_keys_and_node_id(cb2); + { struct utun_config* cfa = parse_config(ca2); g_nid_a2 = cfa->global.my_node_id; free_config(cfa); + struct utun_config* cfb = parse_config(cb2); g_nid_b2 = cfb->global.my_node_id; free_config(cfb); } + char *r0 = gv(ca2,"priv"), *p0 = gv(ca2,"pub"); + char *r1 = gv(cb2,"priv"), *p1 = gv(cb2,"pub"); + wf(ca2, "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.98.0.1/24\ntun_ifname=tun97\n" + "db_path=%s\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", r0, p0, dbdir_a, pa2); + wf(cb2, "[global]\nmy_private_key=%s\nmy_public_key=%s\ntun_ip=10.98.0.2/24\ntun_ifname=tun96\n" + "db_path=%s\n[server: s1]\naddr=127.0.0.1:%d\ntype=public\n[allowed_keys]\nallow_all=1\n", r1, p1, dbdir_b, pb2); + u_free(r0); u_free(p0); u_free(r1); u_free(p1); + /* fake unreachable node C — random id, random pubkey, dead port 55555 */ + g_nid_c = 0xC0DEC0DEC0DEC0DEULL; +} +static void cleanup_db_test(void) { test_unlink(ca2); test_unlink(cb2); test_rmdir(tdir2); } + +/* ---- main ---- */ int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); - utun_instance_set_tun_init_enabled(0); setup(); + utun_instance_set_tun_init_enabled(0); + + /* ===== Phase 1: existing tests with client link ===== */ + setup(); ua = uasync_create(); g_a = utun_instance_create(ua, ca); g_b = utun_instance_create(ua, cb); - if (!g_a || !g_b) goto done; + if (!g_a || !g_b) goto phase1_cleanup; utun_instance_init(g_a); utun_instance_init(g_b); uasync_call_soon(ua, NULL, test1); - ttimer = uasync_set_timeout(ua, TIMEOUT_TB, NULL, to_cb, "to"); + ttimer = uasync_set_timeout(ua, TIMEOUT_TB, NULL, to_cb, "to1"); { int el = 0; while (!result && el < TIMEOUT_TB + 5000) { uasync_poll(ua, POLL_MS); el += POLL_MS; } } - fprintf(stderr, "final result=%d\n", result); fflush(stderr); -done: - if (ttimer) uasync_cancel_timeout(ua, ttimer); - if (g_a) { g_a->running = 0; utun_instance_destroy(g_a); } - if (g_b) { g_b->running = 0; utun_instance_destroy(g_b); } + fprintf(stderr, "phase1 result=%d\n", result); fflush(stderr); +phase1_cleanup: + if (ttimer) { uasync_cancel_timeout(ua, ttimer); ttimer = NULL; } + if (g_a) { g_a->running = 0; utun_instance_destroy(g_a); g_a = NULL; } + if (g_b) { g_b->running = 0; utun_instance_destroy(g_b); g_b = NULL; } if (ua) { uasync_destroy(ua, 0); ua = NULL; } cleanup(); + + if (result == 2) return 1; + result = 0; cdone2 = 0; cresult2 = 0; + + /* ===== Phase 2: DB-loaded node tests ===== */ + setup_db_test(); + ua2 = uasync_create(); + g_a2 = utun_instance_create(ua2, ca2); g_b2 = utun_instance_create(ua2, cb2); + if (!g_a2 || !g_b2) goto done; + utun_instance_init(g_a2); utun_instance_init(g_b2); + + /* Insert B2 and fake C into A2's SQLite (B2 listens on pb2) */ + { + uint8_t b2_pubkey[32], c_pubkey[32]; + memset(c_pubkey, 0xCC, 32); + memcpy(b2_pubkey, g_b2->my_keys.public_key, 32); + uint8_t ip[4] = {127, 0, 0, 1}; + if (db_insert_node(g_a2->topo_sqlite_db, g_nid_b2, b2_pubkey, "B2", ip, pb2) != 0) + { fprintf(stderr, "FAIL: db_insert B2\n"); result = 2; goto done; } + if (db_insert_node(g_a2->topo_sqlite_db, g_nid_c, c_pubkey, "C", ip, 55555) != 0) + { fprintf(stderr, "FAIL: db_insert C\n"); result = 2; goto done; } + } + conn_mgr_set_direct_timeout_ms(g_a2->conn_mgr, DB_TEST_TIMEOUT_MS); + + uasync_call_soon(ua2, NULL, test3); + ttimer = uasync_set_timeout(ua2, DB_TEST_TIMEOUT_MS * 10 + 100000, NULL, to_cb, "to2"); + { int el = 0; while (!result && el < TIMEOUT_TB + 5000) { uasync_poll(ua2, POLL_MS); el += POLL_MS; } } + fprintf(stderr, "phase2 result=%d\n", result); fflush(stderr); + +done: + if (ttimer) { uasync_cancel_timeout(ua2, ttimer); ttimer = NULL; } + if (g_a2) { g_a2->running = 0; utun_instance_destroy(g_a2); g_a2 = NULL; } + if (g_b2) { g_b2->running = 0; utun_instance_destroy(g_b2); g_b2 = NULL; } + if (ua2) { uasync_destroy(ua2, 0); ua2 = NULL; } + cleanup_db_test(); return (result == 1) ? 0 : 1; }