Browse Source
- conn_mgr_cb_t: 5 параметров (h, node_id, group_id, event, arg) - conn_mgr_open_invite: единая функция (ni=NULL — обычное, ni!=NULL — invite) - CONN_EVENT_JOIN для успешного invite - struct CONN_MGR_HANDLE: добавлен group_id - bind/unbind в conn_mgr_init/destroy (один на инстанс, перезапись безвредна) - utun_instance_destroy: conn_mgr_destroy до закрытия сокетов - cm_ncd_callback: занулять ncd_handle при DOWN, защита от вызова после destroy - topo_group_connect: свой трекинг handle'ов, conn_mgr_close для отмены - обновлены: topo_recovery, media_download, media_delivery, chat_sync, тестыtopo_upd
evgeny
2 months ago
17 changed files with 1643 additions and 2358 deletions
@ -0,0 +1,909 @@
|
||||
/* conn_mgr_core.c — init/destroy, open/close/handles, DIRECT, REVERSE, invite */ |
||||
#include "conn_mgr_priv.h" |
||||
|
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
#ifdef _WIN32 |
||||
#include <winsock2.h> |
||||
#include <ws2tcpip.h> |
||||
#else |
||||
#include <arpa/inet.h> |
||||
#endif |
||||
#include "../lib/platform_compat.h" |
||||
#include "../lib/debug_config.h" |
||||
#include "../lib/mem.h" |
||||
#include "../lib/u_async.h" |
||||
#include "utun_instance.h" |
||||
#include "etcp.h" |
||||
#include "etcp_connections.h" |
||||
#include "etcp_router.h" |
||||
#include "config_parser.h" |
||||
#include "topo_node.h" |
||||
#include "topo_node_sqlite.h" |
||||
#include "topo_group.h" |
||||
#include "../transport_layer/stcp_link.h" |
||||
#include "route_ping.h" |
||||
|
||||
/* ═══════ forward-декларации ═══════ */ |
||||
|
||||
void conn_mgr_router_recv_handler(struct ETCP_CONN* conn, struct ll_entry* entry); |
||||
static void cm_handle_invite_info_req(struct CONN_MGR* mgr, struct ETCP_CONN* conn, const uint8_t* data, size_t len); |
||||
static void cm_handle_invite_info_resp(struct CONN_MGR* mgr, const uint8_t* data, size_t len); |
||||
|
||||
/* ═══════ утилиты ═══════ */ |
||||
|
||||
static struct CONN_MGR_HANDLE* cm_handle_new(struct CONN_MGR_ENTRY* entry, uint64_t group_id, |
||||
conn_mgr_cb_t cb, void* cb_arg) { |
||||
struct CONN_MGR_HANDLE* h = u_calloc(1, sizeof(*h)); |
||||
if (!h) return NULL; |
||||
h->node_id = entry->node_id; h->group_id = group_id; h->cb = cb; h->cb_arg = cb_arg; |
||||
h->entry = entry; h->next = entry->handles; entry->handles = h; |
||||
return h; |
||||
} |
||||
|
||||
static struct CONN_MGR_HANDLE* cm_invite_handle_new(conn_mgr_cb_t cb, void* cb_arg, uint64_t node_id, uint64_t group_id) { |
||||
struct CONN_MGR_HANDLE* h = u_calloc(1, sizeof(*h)); |
||||
if (!h) return NULL; |
||||
h->node_id = node_id; h->group_id = group_id; h->cb = cb; h->cb_arg = cb_arg; |
||||
return h; |
||||
} |
||||
|
||||
static void cm_deliver_up_cb(void* arg) { |
||||
struct CONN_MGR_HANDLE* h = (struct CONN_MGR_HANDLE*)arg; |
||||
if (h->cb) h->cb(h, h->node_id, h->entry->mgr->group->group_id, CONN_EVENT_UP, h->cb_arg); |
||||
} |
||||
|
||||
/* Рассылает событие ВСЕМ handle'ам entry. Одно соединение — много слушателей:
|
||||
* каждый conn_mgr_open создал свой handle, все получают UP/DOWN/TIMEOUT. */ |
||||
void cm_deliver_event(struct CONN_MGR_ENTRY* entry, enum conn_mgr_event event) { |
||||
uint64_t gid = entry->mgr->group->group_id; |
||||
struct CONN_MGR_HANDLE* h = entry->handles; |
||||
while (h) { struct CONN_MGR_HANDLE* next = h->next; if (h->cb) h->cb(h, h->node_id, gid, event, h->cb_arg); h = next; } |
||||
} |
||||
|
||||
/* ═══════ NAT / адресные утилиты ═══════ */ |
||||
|
||||
/* Проверяет NAT-совместимость нашего сокета с целевым: возвращает 1 (публичный/EIM —
|
||||
* можно пробить), 2 (локальная сеть — оба PRIVATE), 0 (несовместим). Используется |
||||
* в DIRECT фазе чтобы не создавать заведомо непробиваемые линки. */ |
||||
int cm_nat_compatible(struct ETCP_SOCKET* our, uint8_t tcfg, uint8_t tnat) { |
||||
uint8_t tt = tnat >= NAT_VERIFIED_UNKNOWN ? CFG_SERVER_TYPE_PUBLIC : tcfg; |
||||
uint8_t ot = our->type; |
||||
if (ot == CFG_SERVER_TYPE_PUBLIC || ot == CFG_SERVER_TYPE_UNKNOWN) return 1; |
||||
if (ot == CFG_SERVER_TYPE_NAT && our->nat_type == NAT_VERIFIED_EIM && tt == CFG_SERVER_TYPE_PUBLIC) return 1; |
||||
if (ot == CFG_SERVER_TYPE_NAT && our->nat_type == NAT_VERIFIED_EIM && tt == CFG_SERVER_TYPE_NAT) return 1; |
||||
if ((ot == CFG_SERVER_TYPE_PRIVATE || ot == CFG_SERVER_TYPE_LOCAL) && tt == CFG_SERVER_TYPE_PRIVATE) return 2; |
||||
return 0; |
||||
} |
||||
|
||||
#define CM_V6_LL 0 |
||||
#define CM_V6_LOC 1 |
||||
#define CM_V6_DIR 2 |
||||
#define CM_V6_OTH 3 |
||||
|
||||
/* Классификация IPv6-адреса: link-local (fe80), локальный ULA (fc/fd),
|
||||
* глобальный/прямой, другое (loopback, multicast, ::). */ |
||||
uint8_t cm_classify_v6_addr(const uint8_t addr[16]) { |
||||
if (addr[0] == 0xfe && (addr[1] & 0xc0) == 0x80) return CM_V6_LL; |
||||
if (addr[0] == 0xfc || addr[0] == 0xfd) return CM_V6_LOC; |
||||
if (addr[0] == 0xff) return CM_V6_OTH; |
||||
{ static const uint8_t z[16]; if (!memcmp(addr, z, 16)) return CM_V6_OTH; } |
||||
{ static const uint8_t lb[16]={0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1}; if (!memcmp(addr, lb, 16)) return CM_V6_OTH; } |
||||
return CM_V6_DIR; |
||||
} |
||||
|
||||
uint8_t cm_sock_v6_classify(const struct ETCP_SOCKET* s) { |
||||
if (s->local_addr.ss_family != AF_INET6) return CM_V6_OTH; |
||||
return cm_classify_v6_addr(((const struct sockaddr_in6*)&s->local_addr)->sin6_addr.s6_addr); |
||||
} |
||||
|
||||
void cm_add_v4_link(struct ETCP_CONN* conn, const uint8_t* addr, uint16_t port, struct ETCP_SOCKET* s) { |
||||
struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET; |
||||
memcpy(&sin.sin_addr.s_addr, addr, 4); sin.sin_port = port; |
||||
struct sockaddr_storage sa; memcpy(&sa, &sin, sizeof(sin)); |
||||
etcp_link_new(conn, s, &sa, 0); |
||||
} |
||||
|
||||
void cm_add_v6_link(struct ETCP_CONN* conn, const uint8_t addr[16], uint16_t port, struct ETCP_SOCKET* s) { |
||||
struct sockaddr_in6 sin6; memset(&sin6, 0, sizeof(sin6)); sin6.sin6_family = AF_INET6; |
||||
memcpy(&sin6.sin6_addr, addr, 16); sin6.sin6_port = port; |
||||
if (cm_classify_v6_addr(addr) == CM_V6_LL) { struct sockaddr_in6* our = (struct sockaddr_in6*)&s->local_addr; sin6.sin6_scope_id = our->sin6_scope_id; } |
||||
struct sockaddr_storage sa; memcpy(&sa, &sin6, sizeof(sin6)); |
||||
etcp_link_new(conn, s, &sa, 0); |
||||
} |
||||
|
||||
/* Проверяет есть ли у узла прямые IP (public/EIM/DIRECT на NAT/INTERFACE адресах) —
|
||||
* можем ли мы достучаться до узла напрямую. Определяет, пойдём в REVERSE или INDIRECT. */ |
||||
int cm_has_direct_ip(struct CONN_MGR* mgr, struct TOPO_GROUP_NODE* nq) { |
||||
if (!nq) return 0; |
||||
struct TOPO_NODE* ni = topo_node_registry_find(mgr->instance->topo_groups, nq->node_id); |
||||
if (!ni) return 0; |
||||
const uint8_t pub_conf[2] = {CFG_SERVER_TYPE_PUBLIC, CFG_SERVER_TYPE_UNKNOWN}; |
||||
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) { |
||||
if (a->type != TOPO_ADDR_NAT && a->type != TOPO_ADDR_INTERFACE) continue; |
||||
for (const struct TOPO_SOCKMETA4* m = ni->v4_sock_meta; m; m = m->next) |
||||
if (m->id == a->socket_id && (m->config_type == pub_conf[0] || m->config_type == pub_conf[1] |
||||
|| m->nat_type == NAT_VERIFIED_EIM || m->nat_type == NAT_VERIFIED_DIRECT)) return 1; |
||||
} |
||||
for (const struct TOPO_ADDR6* a6 = ni->v6_addrs; a6; a6 = a6->next) { |
||||
if ((a6->type != TOPO_ADDR_NAT && a6->type != TOPO_ADDR_INTERFACE) || cm_classify_v6_addr(a6->addr) != CM_V6_DIR) continue; |
||||
for (const struct TOPO_SOCKMETA6* m = ni->v6_sock_meta; m; m = m->next) |
||||
if (m->id == a6->socket_id && (m->config_type == pub_conf[0] || m->config_type == pub_conf[1] |
||||
|| m->nat_type == NAT_VERIFIED_EIM || m->nat_type == NAT_VERIFIED_DIRECT)) return 1; |
||||
} |
||||
return 0; |
||||
} |
||||
|
||||
int cm_has_local_addr(struct CONN_MGR* mgr, struct TOPO_GROUP_NODE* nq) { |
||||
if (!nq) return 0; |
||||
struct TOPO_NODE* ni = topo_node_registry_find(mgr->instance->topo_groups, nq->node_id); |
||||
if (!ni) return 0; |
||||
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) { |
||||
if (a->type != TOPO_ADDR_INTERFACE) continue; |
||||
for (const struct TOPO_SOCKMETA4* m = ni->v4_sock_meta; m; m = m->next) |
||||
if (m->id == a->socket_id && (m->config_type == CFG_SERVER_TYPE_PRIVATE || m->config_type == CFG_SERVER_TYPE_LOCAL)) return 1; |
||||
} |
||||
for (const struct TOPO_ADDR6* a6 = ni->v6_addrs; a6; a6 = a6->next) { |
||||
if (a6->type != TOPO_ADDR_INTERFACE) continue; |
||||
uint8_t cls = cm_classify_v6_addr(a6->addr); |
||||
if (cls != CM_V6_LL && cls != CM_V6_LOC) continue; |
||||
for (const struct TOPO_SOCKMETA6* m = ni->v6_sock_meta; m; m = m->next) |
||||
if (m->id == a6->socket_id && (m->config_type == CFG_SERVER_TYPE_PRIVATE || m->config_type == CFG_SERVER_TYPE_LOCAL)) return 1; |
||||
} |
||||
return 0; |
||||
} |
||||
|
||||
/* ═══════ entries ═══════ */ |
||||
|
||||
struct CONN_MGR_ENTRY* cm_find_entry(struct CONN_MGR* mgr, uint64_t node_id) { |
||||
for (size_t i = 0; i < mgr->entry_count; i++) |
||||
if (mgr->entries[i].node_id == node_id && mgr->entries[i].state != CONN_MGR_STATE_DISCONNECTED) |
||||
return &mgr->entries[i]; |
||||
return NULL; |
||||
} |
||||
|
||||
struct CONN_MGR_ENTRY* cm_ensure_entry(struct CONN_MGR* mgr, uint64_t node_id) { |
||||
struct CONN_MGR_ENTRY* e = cm_find_entry(mgr, node_id); |
||||
if (e) return e; |
||||
if (mgr->entry_count >= mgr->entry_capacity) { |
||||
size_t cap = mgr->entry_capacity * 2; |
||||
struct CONN_MGR_ENTRY* ne = u_realloc(mgr->entries, cap * sizeof(struct CONN_MGR_ENTRY)); |
||||
if (!ne) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: realloc entries failed"); return NULL; } |
||||
memset(ne + mgr->entry_capacity, 0, (cap - mgr->entry_capacity) * sizeof(struct CONN_MGR_ENTRY)); |
||||
mgr->entries = ne; mgr->entry_capacity = cap; |
||||
} |
||||
struct CONN_MGR_ENTRY* new_e = &mgr->entries[mgr->entry_count++]; |
||||
memset(new_e, 0, sizeof(*new_e)); new_e->node_id = node_id; new_e->mgr = mgr; |
||||
return new_e; |
||||
} |
||||
|
||||
void cm_entry_cleanup(struct CONN_MGR_ENTRY* entry) { |
||||
if (!entry || entry->state == CONN_MGR_STATE_DISCONNECTED) return; |
||||
if (entry->idle_timer) { uasync_cancel_timeout(entry->mgr->instance->ua, entry->idle_timer); entry->idle_timer = NULL; } |
||||
if (entry->main.timer) { uasync_cancel_timeout(entry->mgr->instance->ua, entry->main.timer); entry->main.timer = NULL; } |
||||
{ struct cm_exchange_pending** pp = &entry->mgr->exchange_pending; |
||||
while (*pp) { |
||||
if ((*pp)->entry == entry) { |
||||
if ((*pp)->timer && (*pp)->timer != entry->main.timer) |
||||
{ uasync_cancel_timeout(entry->mgr->instance->ua, (*pp)->timer); (*pp)->timer = NULL; } |
||||
struct cm_exchange_pending* ep = *pp; *pp = ep->next; u_free(ep); |
||||
} else pp = &(*pp)->next; |
||||
} |
||||
} |
||||
if (entry->ncd_handle) { node_conn_direct_close(entry->ncd_handle); entry->ncd_handle = NULL; } |
||||
entry->handles = NULL; |
||||
cm_clear_nodeinfo(entry->mgr, entry->node_id); |
||||
entry->state = CONN_MGR_STATE_DISCONNECTED; entry->conn_type = CONN_TYPE_NONE; |
||||
} |
||||
|
||||
void cm_clear_nodeinfo(struct CONN_MGR* mgr, uint64_t node_id) { |
||||
struct TOPO_GROUP_NODE* 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; |
||||
} |
||||
|
||||
void cm_update_nodeinfo(struct CONN_MGR_ENTRY* entry) { |
||||
if (!entry->mgr->group) return; |
||||
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(entry->mgr->group, entry->node_id); |
||||
if (!nq) return; |
||||
nq->conn_mgr_type = entry->conn_type; |
||||
memcpy(nq->conn_mgr_intermediaries, entry->intermediaries, sizeof(entry->intermediaries)); |
||||
nq->conn_mgr_intermediariy_count = entry->intermediariy_count; |
||||
} |
||||
|
||||
void cm_cleanup_db_node(struct CONN_MGR_ENTRY* entry) { |
||||
if (!entry->db_loaded) return; |
||||
struct TOPO_GROUP* group = entry->mgr->group; |
||||
struct TOPO_GROUP_NODE* 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_nodeq_free_group_fields(entry->mgr->instance->topo_groups, nq); |
||||
queue_entry_free(&nq->ll); |
||||
} |
||||
|
||||
/* ═══════ NCD коллбэки ═══════ */ |
||||
|
||||
/* NCD события → conn_mgr для DIRECT/REVERSE фаз:
|
||||
* UP — соединение поднялось, помечаем CONNECTED, доставляем UP всем handle'ам. |
||||
* TIMEOUT — NCD таймаут истёк, переходим к REVERSE или INDIRECT, либо фейлим db_node. |
||||
* DOWN — соединение упало, доставляем DOWN. */ |
||||
void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void* arg) { |
||||
struct CONN_MGR_ENTRY* entry = (struct CONN_MGR_ENTRY*)arg; (void)ncd_h; |
||||
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return; |
||||
switch (ncd_ev) { |
||||
case NCD_EVENT_UP: |
||||
if (entry->state == CONN_MGR_STATE_CONNECTED) return; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: NCD UP for 0x%016llx", (unsigned long long)entry->node_id); |
||||
if (entry->main.timer) { uasync_cancel_timeout(entry->mgr->instance->ua, entry->main.timer); entry->main.timer = NULL; } |
||||
entry->main_connect_state = CM_TRY_OK; entry->conn_type = CONN_TYPE_DIRECT; |
||||
entry->state = CONN_MGR_STATE_CONNECTED; |
||||
cm_update_nodeinfo(entry); cm_deliver_event(entry, CONN_EVENT_UP); |
||||
break; |
||||
case NCD_EVENT_TIMEOUT: |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: NCD TIMEOUT for 0x%016llx", (unsigned long long)entry->node_id); |
||||
entry->ncd_handle = NULL; |
||||
if (entry->db_loaded) { |
||||
entry->main_connect_state = CM_TRY_FAILED; |
||||
cm_cleanup_db_node(entry); cm_deliver_event(entry, CONN_EVENT_TIMEOUT); |
||||
return; |
||||
} |
||||
entry->main_connect_state = CM_TRY_FAILED; |
||||
{ struct TOPO_GROUP* g = entry->mgr->group; |
||||
struct TOPO_GROUP_NODE* t = topo_node_find_by_id(g, entry->node_id); |
||||
if (entry->local_scan_state != CM_TRY_OK) { |
||||
if (cm_has_direct_ip(entry->mgr, g->local_node) && !(t && cm_has_direct_ip(entry->mgr, t))) |
||||
{ cm_start_phase_reverse(entry); return; } |
||||
cm_start_phase_indirect(entry); return; |
||||
} |
||||
} |
||||
if (entry->local_scan_state == CM_TRY_FAILED) cm_deliver_event(entry, CONN_EVENT_TIMEOUT); |
||||
break; |
||||
case NCD_EVENT_DOWN: |
||||
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: NCD DOWN for 0x%016llx", (unsigned long long)entry->node_id); |
||||
entry->ncd_handle = NULL; |
||||
cm_deliver_event(entry, CONN_EVENT_DOWN); |
||||
break; |
||||
} |
||||
} |
||||
|
||||
/* NCD → invite: UP (отправляем INVITE_INFO_REQ для проверки членства),
|
||||
* TIMEOUT/DOWN (фейлим invite с CONN_EVENT_TIMEOUT). */ |
||||
void cm_invite_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void* arg) { |
||||
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg; |
||||
switch (ncd_ev) { |
||||
case NCD_EVENT_UP: |
||||
if (inv->state != CM_INVITE_CONNECTING) return; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: invite INIT OK node=0x%016llx", (unsigned long long)inv->node_id); |
||||
inv->state = CM_INVITE_WAIT_INFO; |
||||
{ struct CM_INVITE_REQ req; memset(&req, 0, sizeof(req)); |
||||
req.cmd = ETCP_RT_ID_CONN_MGR; req.subcmd = CM_SUBCMD_INVITE_INFO_REQ; req.group_id = inv->mgr->group->group_id; |
||||
struct ll_entry* qe = queue_entry_new(0); |
||||
if (!qe) { cm_invite_fail(inv); return; } |
||||
qe->dgram = u_malloc(sizeof(req)); if (!qe->dgram) { queue_entry_free(qe); cm_invite_fail(inv); return; } |
||||
memcpy(qe->dgram, &req, sizeof(req)); qe->len = sizeof(req); |
||||
etcp_send(node_conn_direct_get_conn(ncd_h), qe); |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: sent INVITE_INFO_REQ to 0x%016llx group=0x%016llx", |
||||
(unsigned long long)inv->node_id, (unsigned long long)inv->mgr->group->group_id); |
||||
} |
||||
break; |
||||
case NCD_EVENT_TIMEOUT: |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: invite connect timeout node=0x%016llx", (unsigned long long)inv->node_id); |
||||
cm_invite_fail(inv); break; |
||||
case NCD_EVENT_DOWN: |
||||
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: invite conn down node=0x%016llx", (unsigned long long)inv->node_id); |
||||
if (inv->state == CM_INVITE_CONNECTING) cm_invite_fail(inv); |
||||
break; |
||||
} |
||||
} |
||||
|
||||
/* ═══════ init / destroy ═══════ */ |
||||
|
||||
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 = group->instance; mgr->group = group; |
||||
mgr->entry_capacity = 8; mgr->direct_timeout_ms = CONN_MGR_CONNECT_DIRECT_TIMEOUT_MS; |
||||
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; } |
||||
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; |
||||
/* Первый conn_mgr биндит etcp_router (остальные перезаписывают тот же коллбэк — безвредно) */ |
||||
etcp_router_bind(mgr->instance, ETCP_RT_ID_CONN_MGR, conn_mgr_router_recv_handler); |
||||
etcp_bind(mgr->instance, ETCP_RT_ID_CONN_MGR, conn_mgr_router_recv_handler); |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: initialized for group %016llx", (unsigned long long)group->group_id); |
||||
return mgr; |
||||
} |
||||
|
||||
void conn_mgr_destroy(struct CONN_MGR* mgr) { |
||||
if (!mgr) return; |
||||
if (mgr->bg_ping_timer) { uasync_cancel_timeout(mgr->instance->ua, mgr->bg_ping_timer); mgr->bg_ping_timer = NULL; } |
||||
if (mgr->candidate_ping_timer) { uasync_cancel_timeout(mgr->instance->ua, mgr->candidate_ping_timer); mgr->candidate_ping_timer = NULL; } |
||||
/* Отвязываемся от etcp_router (если ещё не отвязаны) */ |
||||
etcp_router_unbind(mgr->instance, ETCP_RT_ID_CONN_MGR); |
||||
etcp_unbind(mgr->instance, ETCP_RT_ID_CONN_MGR); |
||||
for (size_t i = 0; i < mgr->entry_count; i++) cm_entry_cleanup(&mgr->entries[i]); |
||||
while (mgr->invite_list) cm_invite_fail(mgr->invite_list); |
||||
while (mgr->exchange_pending) { |
||||
struct cm_exchange_pending* ep = mgr->exchange_pending; mgr->exchange_pending = ep->next; |
||||
if (ep->timer) uasync_cancel_timeout(mgr->instance->ua, ep->timer); |
||||
u_free(ep); |
||||
} |
||||
while (mgr->reverse_pending) { struct cm_reverse_pending* rp = mgr->reverse_pending; mgr->reverse_pending = rp->next; u_free(rp); } |
||||
u_free(mgr->entries); mgr->initialized = 0; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: destroyed, entries=%zu", (size_t)mgr->entry_count); |
||||
u_free(mgr); |
||||
} |
||||
|
||||
/* ═══════ публичное API ═══════ */ |
||||
|
||||
int conn_mgr_open(struct CONN_MGR* mgr, uint64_t node_id, uint32_t idle_timeout_ms, |
||||
conn_mgr_cb_t cb, void* cb_arg, struct CONN_MGR_HANDLE** out_handle) { |
||||
if (out_handle) *out_handle = NULL; |
||||
if (!mgr) { if (cb) cb(NULL, node_id, 0, CONN_EVENT_TIMEOUT, cb_arg); return -1; } |
||||
struct TOPO_GROUP* group = mgr->group; |
||||
if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "BGP not initialized"); return -1; } |
||||
uint64_t gid = group->group_id; |
||||
struct TOPO_GROUP_NODE* 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_store(mgr->instance->topo_groups, ni); |
||||
struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_GROUP_NODE)); |
||||
if (qe) { |
||||
target = (struct TOPO_GROUP_NODE*)qe; |
||||
memset((uint8_t*)target + sizeof(struct ll_entry), 0, sizeof(*target) - sizeof(struct ll_entry)); |
||||
target->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_GROUP_NODE 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 -1; } |
||||
|
||||
struct CONN_MGR_ENTRY* entry = cm_find_entry(mgr, node_id); |
||||
|
||||
if (entry && entry->state == CONN_MGR_STATE_CONNECTED) { |
||||
struct CONN_MGR_HANDLE* h = cm_handle_new(entry, gid, cb, cb_arg); |
||||
if (!h) return -1; if (out_handle) *out_handle = h; |
||||
uasync_call_soon(mgr->instance->ua, h, cm_deliver_up_cb); return 0; |
||||
} |
||||
if (entry && entry->state == CONN_MGR_STATE_CONNECTING) { |
||||
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "node 0x%016llx already connecting, adding handle", (unsigned long long)node_id); |
||||
struct CONN_MGR_HANDLE* h = cm_handle_new(entry, gid, cb, cb_arg); |
||||
if (!h) return -1; if (out_handle) *out_handle = h; return 0; |
||||
} |
||||
|
||||
struct ETCP_CONN* existing = topo_group_find_conn_for_node(group, node_id); |
||||
if (!existing) existing = instance_find_conn(mgr->instance, node_id); |
||||
if (existing && existing->peer_node_id == node_id && existing->links) { |
||||
struct ETCP_LINK* l = existing->links; |
||||
while (l) { if (l->link_state == 3 && l->initialized) break; l = l->next; } |
||||
if (l) { |
||||
entry = cm_ensure_entry(mgr, node_id); |
||||
if (!entry) return -1; |
||||
entry->state = CONN_MGR_STATE_CONNECTED; entry->conn_type = CONN_TYPE_DIRECT; |
||||
entry->idle_timeout_ms = idle_timeout_ms; entry->last_traffic_tb = get_time_tb(); |
||||
{ struct TOPO_NODE* ni = topo_node_registry_find(mgr->instance->topo_groups, node_id); |
||||
struct TOPO_NODE ni_e; memset(&ni_e, 0, sizeof(ni_e)); ni_e.node_id = node_id; |
||||
if (ni) memcpy(ni_e.public_key, ni->public_key, SC_PUBKEY_SIZE); |
||||
node_conn_direct_open_node(mgr->instance, node_id, cm_ncd_callback, entry, &entry->ncd_handle, &ni_e, NULL); } |
||||
struct CONN_MGR_HANDLE* h = cm_handle_new(entry, gid, cb, cb_arg); |
||||
if (!h) { cm_entry_cleanup(entry); return -1; } if (out_handle) *out_handle = h; |
||||
cm_update_nodeinfo(entry); uasync_call_soon(mgr->instance->ua, h, cm_deliver_up_cb); return 0; |
||||
} |
||||
} |
||||
|
||||
entry = cm_ensure_entry(mgr, node_id); |
||||
if (!entry) return -1; |
||||
entry->state = CONN_MGR_STATE_CONNECTING; |
||||
struct CONN_MGR_HANDLE* h = cm_handle_new(entry, gid, cb, cb_arg); |
||||
if (!h) { cm_entry_cleanup(entry); return -1; } if (out_handle) *out_handle = h; |
||||
entry->idle_timeout_ms = idle_timeout_ms; entry->last_traffic_tb = get_time_tb(); |
||||
entry->local_scan_state = CM_TRY_NONE; entry->main_connect_state = CM_TRY_NONE; |
||||
entry->db_loaded = (uint8_t)loaded_from_db; |
||||
if (cm_has_local_addr(mgr, group->local_node) && cm_has_local_addr(mgr, target)) |
||||
{ entry->local_scan_state = CM_TRY_PENDING; cm_start_local_scan(entry); } |
||||
entry->main_connect_state = CM_TRY_PENDING; cm_start_phase_direct(entry); |
||||
return 0; |
||||
} |
||||
|
||||
void conn_mgr_close(struct CONN_MGR_HANDLE* h) { |
||||
if (!h) return; |
||||
struct CONN_MGR_ENTRY* entry = h->entry; |
||||
if (!entry) { u_free(h); return; } |
||||
{ struct CONN_MGR_HANDLE** pp = &entry->handles; |
||||
while (*pp) { if (*pp == h) { *pp = h->next; break; } pp = &(*pp)->next; } |
||||
} |
||||
if (entry->handles) { u_free(h); return; } |
||||
if (entry->state == CONN_MGR_STATE_CONNECTED && (entry->conn_type == CONN_TYPE_DIRECT || entry->conn_type == CONN_TYPE_REVERSE)) { |
||||
/* DISCONNECT через etcp_router — conn_mgr уровень */ |
||||
struct CM_DISCONNECT pkt; memset(&pkt, 0, sizeof(pkt)); |
||||
pkt.cmd = ETCP_RT_ID_CONN_MGR; pkt.subcmd = CM_SUBCMD_DISCONNECT; pkt.node_id = entry->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(entry->mgr->instance, entry->mgr->group->group_id, entry->node_id, qe, 1); } |
||||
/* NCD CLOSE — транспортный уровень (отправит CLOSE/KEEP_ALIVE через node_conn_direct) */ |
||||
} |
||||
cm_entry_cleanup(entry); u_free(h); |
||||
} |
||||
|
||||
struct ETCP_CONN* conn_mgr_get_conn(struct CONN_MGR_HANDLE* h) { |
||||
if (!h || !h->entry || !h->entry->ncd_handle) return NULL; |
||||
return node_conn_direct_get_conn(h->entry->ncd_handle); |
||||
} |
||||
|
||||
int conn_mgr_send(struct CONN_MGR_HANDLE* h, struct ll_entry* e) { |
||||
if (!h || !e) return -1; |
||||
struct CONN_MGR_ENTRY* entry = h->entry; |
||||
if (!entry || entry->state != CONN_MGR_STATE_CONNECTED) { queue_entry_free(e); return -1; } |
||||
entry->last_traffic_tb = get_time_tb(); |
||||
struct ETCP_ROUTER_CONN* r = etcp_router_conn_get(entry->mgr->instance, entry->mgr->group->group_id, entry->node_id, ETCP_RT_ID_CONN_MGR); |
||||
if (!r) { queue_entry_free(e); return -1; } |
||||
int ret = etcp_router_conn_send(r, e->dgram, e->len); queue_entry_free(e); return ret; |
||||
} |
||||
|
||||
/* ═══════ локальное сканирование ═══════ */ |
||||
|
||||
static void cm_ping_cb(int success, uint16_t rtt, void* arg, uint64_t nonce, const uint8_t* resp_data, size_t resp_data_len); |
||||
static void cm_ping_cb_impl(int success, uint16_t rtt, void* arg, uint64_t nonce, const uint8_t* resp_data, size_t resp_data_len); |
||||
|
||||
struct cm_ping_ctx { struct CONN_MGR_ENTRY* entry; struct sockaddr_storage addr; struct ETCP_SOCKET* sock; uint8_t phase, attempt; }; |
||||
|
||||
/* LAN broadcast ping для обнаружения узла в локальной сети. Если ответит —
|
||||
* соединение готово быстрее чем через интернет, без NAT. Запускается |
||||
* параллельно с DIRECT фазой. */ |
||||
void cm_start_local_scan(struct CONN_MGR_ENTRY* entry) { |
||||
struct TOPO_GROUP* group = entry->mgr->group; |
||||
struct TOPO_GROUP_NODE* target = topo_node_find_by_id(group, entry->node_id); |
||||
struct TOPO_NODE* ni = target ? topo_node_registry_find(entry->mgr->instance->topo_groups, target->node_id) : NULL; |
||||
if (!target || !ni) { entry->local_scan_state = CM_TRY_FAILED; return; } |
||||
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) { |
||||
if (a->type != TOPO_ADDR_INTERFACE) continue; |
||||
for (const struct TOPO_SOCKMETA4* m = ni->v4_sock_meta; m; m = m->next) { |
||||
if (m->id != a->socket_id) continue; |
||||
if (m->config_type != CFG_SERVER_TYPE_PRIVATE && m->config_type != CFG_SERVER_TYPE_LOCAL) continue; |
||||
struct ETCP_SOCKET* s = entry->mgr->instance->etcp_sockets; |
||||
while (s) { |
||||
if ((s->type == CFG_SERVER_TYPE_PRIVATE || s->type == CFG_SERVER_TYPE_LOCAL) && s->local_addr.ss_family == AF_INET) { |
||||
struct cm_ping_ctx* ctx = u_calloc(1, sizeof(struct cm_ping_ctx)); |
||||
if (!ctx) continue; |
||||
ctx->entry = entry; ctx->sock = s; ctx->phase = 0; ctx->attempt = 0; |
||||
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); |
||||
memcpy(&ctx->addr, &sin, sizeof(sin)); |
||||
etcp_send_ping_to_socket(entry->mgr->instance, s, ni->public_key, &ctx->addr, CONN_MGR_LOCAL_SCAN_TIMEOUT_MS, cm_ping_cb, ctx, NULL, 0, 0); |
||||
return; |
||||
} |
||||
s = s->next; |
||||
} |
||||
} |
||||
} |
||||
entry->local_scan_state = CM_TRY_FAILED; |
||||
} |
||||
|
||||
static void cm_ping_cb(int success, uint16_t rtt, void* arg, uint64_t nonce, const uint8_t* resp_data, size_t resp_data_len) |
||||
{ cm_ping_cb_impl(success, rtt, arg, nonce, resp_data, resp_data_len); } |
||||
|
||||
static void cm_ping_cb_impl(int success, uint16_t rtt, void* arg, uint64_t nonce, const uint8_t* resp_data, size_t resp_data_len) { |
||||
struct cm_ping_ctx* ctx = (struct cm_ping_ctx*)arg; |
||||
struct CONN_MGR_ENTRY* entry = ctx->entry; (void)nonce; (void)resp_data; (void)resp_data_len; |
||||
if (entry->main_connect_state == CM_TRY_OK) { u_free(ctx); return; } |
||||
if (success && ctx->phase == 0) { |
||||
entry->local_scan_state = CM_TRY_OK; entry->conn_type = CONN_TYPE_DIRECT; entry->state = CONN_MGR_STATE_CONNECTED; |
||||
if (entry->main.timer) { uasync_cancel_timeout(entry->mgr->instance->ua, entry->main.timer); entry->main.timer = NULL; } |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: local scan OK for 0x%016llx rtt=%u", (unsigned long long)entry->node_id, rtt); |
||||
cm_update_nodeinfo(entry); cm_deliver_event(entry, CONN_EVENT_UP); u_free(ctx); return; |
||||
} |
||||
if (++ctx->attempt < CONN_MGR_LOCAL_SCAN_ATTEMPTS && ctx->phase == 0) { |
||||
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(entry->mgr->group, entry->node_id); |
||||
struct TOPO_NODE* pub_ni = nq ? topo_node_registry_find(entry->mgr->instance->topo_groups, nq->node_id) : NULL; |
||||
etcp_send_ping_to_socket(entry->mgr->instance, ctx->sock, pub_ni ? pub_ni->public_key : NULL, |
||||
&ctx->addr, CONN_MGR_LOCAL_SCAN_TIMEOUT_MS, cm_ping_cb, ctx, NULL, 0, 0); |
||||
return; |
||||
} |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: local scan failed for 0x%016llx after %d attempts", |
||||
(unsigned long long)entry->node_id, ctx->attempt); |
||||
entry->local_scan_state = CM_TRY_FAILED; u_free(ctx); |
||||
} |
||||
|
||||
/* ═══════ DIRECT фаза (NCD + ручная NAT-фильтрация линков) ═══════ */ |
||||
|
||||
/* Добавляет IPv4/IPv6 линки к conn. db_loaded: все сокеты без фильтра.
|
||||
* Обычный путь: NAT-фильтрация (cm_nat_compatible) + приоритет NAT > INTERFACE. |
||||
* Если ни одного линка не создалось — закрывает NCD и переходит к REVERSE/INDIRECT. */ |
||||
static void cm_direct_add_links(struct ETCP_CONN* conn, struct CONN_MGR_ENTRY* entry, |
||||
struct TOPO_NODE* ni, int db_loaded) { |
||||
int any = 0; |
||||
/* v4 */ |
||||
if (db_loaded) { |
||||
for (const struct TOPO_ADDR4* a = ni->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; memcpy(&sa,&sin,sizeof(sin)); |
||||
if (etcp_link_new(conn,s,&sa,0)) any=1; } s=s->next; } |
||||
} |
||||
} else { |
||||
uint8_t prio[]={TOPO_ADDR_NAT,TOPO_ADDR_INTERFACE}; |
||||
for (int pr=0;pr<2;pr++) for (const struct TOPO_ADDR4* a=ni->v4_addrs;a;a=a->next) { |
||||
if (a->type!=prio[pr]) continue; |
||||
for (const struct TOPO_SOCKMETA4* m=ni->v4_sock_meta;m;m=m->next) { |
||||
if (m->id!=a->socket_id) continue; |
||||
struct ETCP_SOCKET* s=entry->mgr->instance->etcp_sockets; |
||||
while(s){if(cm_nat_compatible(s,m->config_type,m->nat_type)&&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; memcpy(&sa,&sin,sizeof(sin)); |
||||
if(etcp_link_new(conn,s,&sa,0)) any=1;} s=s->next;} |
||||
} |
||||
} |
||||
} |
||||
/* v6 */ |
||||
for (const struct TOPO_ADDR6* a6=ni->v6_addrs;a6;a6=a6->next) { |
||||
if (!(a6->protocol&TOPO_PROTO_UDP)) continue; |
||||
uint8_t tc=cm_classify_v6_addr(a6->addr); if(tc==CM_V6_OTH) continue; |
||||
struct ETCP_SOCKET* s=entry->mgr->instance->etcp_sockets; |
||||
while(s){if(cm_sock_v6_classify(s)==tc){cm_add_v6_link(conn,a6->addr,a6->port,s);any=1;} s=s->next;} |
||||
} |
||||
|
||||
if (any) return; |
||||
node_conn_direct_close(entry->ncd_handle); entry->ncd_handle=NULL; |
||||
if (db_loaded) { |
||||
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_event(entry, CONN_EVENT_TIMEOUT); |
||||
} else { |
||||
struct TOPO_GROUP* g = entry->mgr->group; |
||||
struct TOPO_GROUP_NODE* t = topo_node_find_by_id(g, entry->node_id); |
||||
if (cm_has_direct_ip(entry->mgr, g->local_node) && !(t && cm_has_direct_ip(entry->mgr, t))) |
||||
{ cm_start_phase_reverse(entry); return; } |
||||
cm_start_phase_indirect(entry); |
||||
} |
||||
} |
||||
|
||||
/* DIRECT фаза: открывает NCD-соединение с пустым ni (0 линков из NCD), затем
|
||||
* вручную добавляет NAT-фильтрованные линки через cm_direct_add_links. |
||||
* NCD управляет таймаутом (2с). При успехе — cm_ncd_callback доставит UP, |
||||
* при таймауте — переключится на REVERSE или INDIRECT. */ |
||||
void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) { |
||||
if (entry->main_connect_state == CM_TRY_OK || entry->local_scan_state == CM_TRY_OK) return; |
||||
struct TOPO_GROUP* group = entry->mgr->group; |
||||
struct TOPO_GROUP_NODE* target = topo_node_find_by_id(group, entry->node_id); |
||||
struct TOPO_NODE* ni = target ? topo_node_registry_find(entry->mgr->instance->topo_groups, target->node_id) : NULL; |
||||
if (!target || !ni) { cm_deliver_event(entry, CONN_EVENT_TIMEOUT); return; } |
||||
|
||||
struct TOPO_NODE ni_e; memset(&ni_e, 0, sizeof(ni_e)); ni_e.node_id = target->node_id; |
||||
memcpy(ni_e.public_key, ni->public_key, SC_PUBKEY_SIZE); |
||||
int r = node_conn_direct_open_node(entry->mgr->instance, entry->node_id, cm_ncd_callback, entry, |
||||
&entry->ncd_handle, &ni_e, NULL); |
||||
if (r == NCD_ERR) { cm_deliver_event(entry, CONN_EVENT_TIMEOUT); return; } |
||||
|
||||
cm_direct_add_links(node_conn_direct_get_conn(entry->ncd_handle), entry, ni, entry->db_loaded); |
||||
} |
||||
|
||||
/* ═══════ REVERSE фаза ═══════ */ |
||||
|
||||
void cm_reverse_init_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event; |
||||
struct cm_reverse_pending* rp = (struct cm_reverse_pending*)arg; |
||||
if (!rp) return; if (!conn) { u_free(rp); return; } |
||||
struct CONN_MGR_ENTRY* entry = rp->entry; |
||||
if (!entry) { DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: reverse ready for 0x%016llx (no entry, freeing)", (unsigned long long)conn->peer_node_id); u_free(rp); return; } |
||||
if (conn->peer_node_id != entry->node_id) { u_free(rp); return; } |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: reverse ready for 0x%016llx", (unsigned long long)entry->node_id); |
||||
if (entry->main.timer) { uasync_cancel_timeout(entry->mgr->instance->ua, entry->main.timer); entry->main.timer = NULL; } |
||||
{ struct cm_reverse_pending** pp = &entry->mgr->reverse_pending; |
||||
while (*pp) { if (*pp == rp) { *pp = rp->next; break; } pp = &(*pp)->next; } |
||||
} |
||||
entry->main_connect_state = CM_TRY_OK; entry->conn_type = CONN_TYPE_REVERSE; entry->state = CONN_MGR_STATE_CONNECTED; |
||||
cm_update_nodeinfo(entry); cm_deliver_event(entry, CONN_EVENT_UP); u_free(rp); |
||||
} |
||||
|
||||
void cm_reverse_timeout_cb(void* arg) { |
||||
struct CONN_MGR_ENTRY* entry = (struct CONN_MGR_ENTRY*)arg; entry->main.timer = NULL; |
||||
{ struct cm_reverse_pending** pp = &entry->mgr->reverse_pending; |
||||
while (*pp) { if ((*pp)->entry == entry) { struct cm_reverse_pending* rp = *pp; *pp = rp->next; u_free(rp); break; } pp = &(*pp)->next; } |
||||
} |
||||
cm_start_phase_indirect(entry); |
||||
} |
||||
|
||||
/* REVERSE фаза: у нас прямой IP, у цели нет. Отправляем DIRECT_REQ с НАШИМИ
|
||||
* адресами цели через BGP-маршрут ("подключись ко мне"). Ждём обратного INIT. |
||||
* Таймаут 15с — при провале переходим к INDIRECT. */ |
||||
void cm_start_phase_reverse(struct CONN_MGR_ENTRY* entry) { |
||||
struct TOPO_GROUP* group = entry->mgr->group; |
||||
struct TOPO_GROUP_NODE* local = group->local_node; |
||||
struct TOPO_NODE* lni = local ? topo_node_registry_find(entry->mgr->instance->topo_groups, local->node_id) : NULL; |
||||
if (!local || !lni) { cm_deliver_event(entry, CONN_EVENT_TIMEOUT); return; } |
||||
uint32_t req_id = ++entry->mgr->next_request_id; entry->main.request_id = req_id; entry->main.phase = 2; |
||||
uint8_t dc = 0; struct { uint8_t t; uint8_t ip[4]; uint16_t port; uint8_t sid; } addrs[8]; |
||||
for (const struct TOPO_ADDR4* a = lni->v4_addrs; a && dc < 8; a = a->next) { |
||||
if (a->type != TOPO_ADDR_NAT && a->type != TOPO_ADDR_INTERFACE) continue; |
||||
for (const struct TOPO_SOCKMETA4* m = lni->v4_sock_meta; m; m = m->next) { |
||||
if (m->id == a->socket_id && (m->config_type == CFG_SERVER_TYPE_PUBLIC || m->config_type == CFG_SERVER_TYPE_UNKNOWN |
||||
|| m->nat_type == NAT_VERIFIED_EIM || m->nat_type == NAT_VERIFIED_DIRECT)) |
||||
{ addrs[dc]=(typeof(addrs[0])){a->type}; memcpy(addrs[dc].ip,a->addr,4); addrs[dc].port=htons(a->port); addrs[dc].sid=a->socket_id; dc++; break; } |
||||
} |
||||
} |
||||
if (!dc) { cm_start_phase_indirect(entry); return; } |
||||
size_t sz = CM_DIRECT_REQ_H_SIZE + (size_t)dc * 8; uint8_t* pkt = u_malloc(sz); |
||||
if (!pkt) { cm_deliver_event(entry, CONN_EVENT_TIMEOUT); return; } |
||||
struct CM_DIRECT_REQ* req = (struct CM_DIRECT_REQ*)pkt; memset(req,0,sizeof(*req)); |
||||
req->cmd=ETCP_RT_ID_CONN_MGR; req->subcmd=CM_SUBCMD_DIRECT_REQ; req->request_id=req_id; req->addr_count=dc; |
||||
uint8_t* p = pkt + CM_DIRECT_REQ_H_SIZE; |
||||
for (uint8_t i=0;i<dc;i++) { *p++=addrs[i].t; memcpy(p,addrs[i].ip,4);p+=4; memcpy(p,&addrs[i].port,2);p+=2; *p++=addrs[i].sid; } |
||||
struct ll_entry* qe = queue_entry_new(0); |
||||
if (!qe) { u_free(pkt); cm_deliver_event(entry, CONN_EVENT_TIMEOUT); return; } |
||||
qe->dgram=pkt; qe->len=(uint16_t)sz; etcp_route_send(entry->mgr->instance, group->group_id, entry->node_id, qe, 1); |
||||
struct cm_reverse_pending* rp = u_calloc(1, sizeof(*rp)); |
||||
if (rp) { rp->request_id=req_id; rp->entry=entry; rp->next=entry->mgr->reverse_pending; entry->mgr->reverse_pending=rp; } |
||||
entry->main.timer = uasync_set_timeout(entry->mgr->instance->ua, CONN_MGR_CONNECT_REVERSE_TIMEOUT_MS*10, |
||||
entry, cm_reverse_timeout_cb, "conn_mgr_reverse"); |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: phase 2 (reverse) sent DIRECT_REQ to 0x%016llx with %u addrs", |
||||
(unsigned long long)entry->node_id, dc); |
||||
} |
||||
|
||||
/* ═══════ обработка DIRECT_REQ (REVERSE входящие) ═══════ */ |
||||
|
||||
/* Принимающая сторона REVERSE: получили DIRECT_REQ от инициатора — открываем
|
||||
* NCD-соединение к нему, добавляем линки по адресам из запроса. При INIT |
||||
* вызывается cm_reverse_init_cb. */ |
||||
void cm_handle_direct_req(struct ETCP_CONN* conn, const uint8_t* data, size_t len) { |
||||
struct CONN_MGR* mgr = conn->instance->conn_mgr; if (!mgr) return; |
||||
struct CM_DIRECT_REQ* req = (struct CM_DIRECT_REQ*)data; |
||||
if (len < CM_DIRECT_REQ_H_SIZE + (size_t)req->addr_count * 8) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: DIRECT_REQ too short"); return; } |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: got DIRECT_REQ from 0x%016llx with %u addrs", |
||||
(unsigned long long)conn->peer_node_id, req->addr_count); |
||||
uint64_t src = conn->peer_node_id; |
||||
struct TOPO_NODE* src_ni = topo_node_registry_find(mgr->instance->topo_groups, src); if (!src_ni) return; |
||||
struct TOPO_NODE nbuf; memset(&nbuf,0,sizeof(nbuf)); nbuf.node_id=src; memcpy(nbuf.public_key,src_ni->public_key,SC_PUBKEY_SIZE); |
||||
struct NODE_CONN_DIRECT* ncd_h = NULL; |
||||
if (node_conn_direct_open_node(mgr->instance, src, NULL, NULL, &ncd_h, &nbuf, NULL) == NCD_ERR) return; |
||||
struct ETCP_CONN* newc = node_conn_direct_get_conn(ncd_h); if (!newc) { node_conn_direct_close(ncd_h); return; } |
||||
struct cm_reverse_pending* rp = u_calloc(1,sizeof(*rp)); |
||||
if (!rp) { node_conn_direct_close(ncd_h); return; } |
||||
rp->request_id=req->request_id; rp->next=mgr->reverse_pending; mgr->reverse_pending=rp; |
||||
|
||||
uint8_t* p = (uint8_t*)data + CM_DIRECT_REQ_H_SIZE; |
||||
for (uint8_t i=0;i<req->addr_count;i++) { |
||||
uint8_t type=*p++; (void)type; uint8_t ip[4]; memcpy(ip,p,4);p+=4; uint16_t port; memcpy(&port,p,2);p+=2; uint8_t sid=*p++; (void)sid; |
||||
struct ETCP_SOCKET* s = conn->instance->etcp_sockets; |
||||
while (s) { if ((s->type==CFG_SERVER_TYPE_PUBLIC||s->type==CFG_SERVER_TYPE_UNKNOWN)&&s->local_addr.ss_family==AF_INET) |
||||
{ cm_add_v4_link(newc, ip, port, s); break; } s=s->next; } |
||||
} |
||||
etcp_conn_add_cbk(newc, cm_reverse_init_cb, rp, ETCP_CBK_EVENT_INIT); |
||||
} |
||||
|
||||
/* ═══════ disconnect / recv ═══════ */ |
||||
|
||||
void cm_handle_disconnect(struct CONN_MGR* mgr, uint64_t node_id) { |
||||
struct CONN_MGR_ENTRY* entry = cm_find_entry(mgr, node_id); if (!entry) return; |
||||
cm_clear_nodeinfo(mgr, node_id); cm_entry_cleanup(entry); |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: remote disconnect from 0x%016llx", (unsigned long long)node_id); |
||||
} |
||||
|
||||
void conn_mgr_router_recv_handler(struct ETCP_CONN* conn, struct ll_entry* entry) { |
||||
if (!entry || !entry->dgram || entry->len < 3) return; |
||||
uint8_t* d = entry->dgram + 1; size_t len = entry->len - 1; uint8_t sub = d[1]; |
||||
struct CONN_MGR* mgr = conn->instance->conn_mgr; |
||||
if (mgr) { struct CONN_MGR_ENTRY* e = cm_find_entry(mgr, conn->peer_node_id); if (e && e->state == CONN_MGR_STATE_CONNECTED) e->last_traffic_tb = get_time_tb(); } |
||||
switch (sub) { |
||||
case CM_SUBCMD_DIRECT_REQ: cm_handle_direct_req(conn, d, len); break; |
||||
case CM_SUBCMD_DIRECT_RESP: break; |
||||
case CM_SUBCMD_INTERM_EXCHANGE_REQ: if (len >= CM_EXCHANGE_REQ_SIZE) cm_handle_interm_exchange_req(conn, (struct CM_EXCHANGE_REQ*)d); break; |
||||
case CM_SUBCMD_INTERM_EXCHANGE_RESP: if (mgr) cm_handle_interm_exchange_resp(mgr, d, len); break; |
||||
case CM_SUBCMD_INTERM_SELECTED: if (mgr) cm_handle_interm_selected(mgr, d, len); break; |
||||
case CM_SUBCMD_DISCONNECT: if (len >= CM_DISCONNECT_SIZE && mgr) cm_handle_disconnect(mgr, ((struct CM_DISCONNECT*)d)->node_id); break; |
||||
case CM_SUBCMD_INVITE_INFO_REQ: if (mgr && len >= sizeof(struct CM_INVITE_REQ)) cm_handle_invite_info_req(mgr, conn, d, len); break; |
||||
case CM_SUBCMD_INVITE_INFO_RESP: if (mgr) cm_handle_invite_info_resp(mgr, d, len); break; |
||||
} |
||||
} |
||||
|
||||
/* ═══════ invite обработчики ═══════ */ |
||||
|
||||
void cm_handle_invite_info_req(struct CONN_MGR* mgr, struct ETCP_CONN* conn, const uint8_t* data, size_t len) { |
||||
if (len < sizeof(struct CM_INVITE_REQ)) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: INVITE_INFO_REQ too short (%zu)", len); return; } |
||||
struct CM_INVITE_REQ* req = (struct CM_INVITE_REQ*)data; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: INVITE_INFO_REQ from 0x%016llx group=0x%016llx", (unsigned long long)conn->peer_node_id, (unsigned long long)req->group_id); |
||||
struct TOPO_GROUP* grp = topo_groups_find(mgr->instance->topo_groups, req->group_id); |
||||
struct TOPO_GROUP* sg = grp ? grp : mgr->group; |
||||
struct TOPO_NODE* ln = sg->local_node ? topo_node_registry_find(sg->instance->topo_groups, sg->local_node->node_id) : NULL; |
||||
const char* nm = ln ? ln->node_name : NULL; size_t nl = nm ? strlen(nm) : 0; if (nl > 255) nl = 255; |
||||
size_t rs = sizeof(struct CM_INVITE_RESP) + nl; |
||||
struct CM_INVITE_RESP* resp = u_calloc(1, rs); if (!resp) return; |
||||
resp->cmd=ETCP_RT_ID_CONN_MGR; resp->subcmd=CM_SUBCMD_INVITE_INFO_RESP; resp->group_id=req->group_id; resp->is_member=1; |
||||
memcpy(resp->ed25519_pubkey, sg->ed25519_public_key, SC_PUBKEY_SIZE); resp->node_name_len=(uint8_t)nl; |
||||
if (nl) memcpy(resp->node_name, nm, nl); |
||||
struct ll_entry* qe = queue_entry_new(0); |
||||
if (qe) { qe->dgram=(uint8_t*)resp; qe->len=(uint16_t)rs; etcp_send(conn, qe); } else u_free(resp); |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: sent INVITE_INFO_RESP to 0x%016llx name_len=%zu", (unsigned long long)conn->peer_node_id, nl); |
||||
} |
||||
|
||||
void cm_handle_invite_info_resp(struct CONN_MGR* mgr, const uint8_t* data, size_t len) { |
||||
if (len < sizeof(struct CM_INVITE_RESP)) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: INVITE_INFO_RESP too short (%zu)", len); return; } |
||||
struct CM_INVITE_RESP* resp = (struct CM_INVITE_RESP*)data; |
||||
if (len < sizeof(struct CM_INVITE_RESP) + (size_t)resp->node_name_len) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: INVITE_INFO_RESP truncated"); return; } |
||||
struct cm_invite_pending* inv = NULL; |
||||
for (struct cm_invite_pending* p = mgr->invite_list; p; p = p->next) if (p->state == CM_INVITE_WAIT_INFO) { inv = p; break; } |
||||
if (!inv) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: INVITE_INFO_RESP — no pending invite in WAIT_INFO"); return; } |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: INVITE_INFO_RESP node=0x%016llx group=0x%016llx member=%d name_len=%u", |
||||
(unsigned long long)inv->node_id, (unsigned long long)resp->group_id, resp->is_member, resp->node_name_len); |
||||
if (!resp->is_member) { cm_invite_fail(inv); return; } |
||||
if (inv->temp_nq) { |
||||
struct TOPO_NODE* ni2 = topo_node_registry_find(inv->mgr->instance->topo_groups, inv->temp_nq->node_id); |
||||
if (ni2) { memcpy(ni2->ed25519_public_key, resp->ed25519_pubkey, SC_PUBKEY_SIZE); |
||||
if (ni2->node_name) { u_free(ni2->node_name); ni2->node_name = NULL; } |
||||
if (resp->node_name_len) ni2->node_name = u_strdup((const char*)resp->node_name); } |
||||
if (inv->mgr->instance->topo_sqlite_db) topo_node_sqlite_node_put(inv->mgr->instance->topo_sqlite_db, inv->mgr->instance->topo_groups, inv->temp_nq, (time_t)(get_time_tb()/10000)); |
||||
} |
||||
if (inv->overall_timer) { uasync_cancel_timeout(inv->mgr->instance->ua, inv->overall_timer); inv->overall_timer = NULL; } |
||||
if (inv->cb) inv->cb(inv->handle, inv->node_id, inv->mgr->group->group_id, CONN_EVENT_JOIN, inv->cb_arg); |
||||
cm_invite_cleanup(inv); |
||||
} |
||||
|
||||
void cm_invite_overall_timeout(void* arg) { |
||||
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg; inv->overall_timer = NULL; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: invite overall timeout node=0x%016llx", (unsigned long long)inv->node_id); |
||||
cm_invite_fail(inv); |
||||
} |
||||
|
||||
/* Очистка invite при ошибке/таймауте: отменяет overall_timer, закрывает TCP-link
|
||||
* и NCD-handle, удаляет временный TOPO_GROUP_NODE, доставляет CONN_EVENT_TIMEOUT |
||||
* через handle и освобождает его. Вызывается также из conn_mgr_destroy. */ |
||||
void cm_invite_fail(struct cm_invite_pending* inv) { |
||||
if (!inv) return; |
||||
if (inv->overall_timer) { uasync_cancel_timeout(inv->mgr->instance->ua, inv->overall_timer); inv->overall_timer = NULL; } |
||||
if (inv->tcp_link) { |
||||
struct ll_entry* e = queue_find_data_by_index(inv->mgr->instance->tcp_connections, (const uint8_t*)&inv->node_id); |
||||
if (e) queue_remove_data(inv->mgr->instance->tcp_connections, e); |
||||
stcp_link_close(inv->tcp_link); inv->tcp_link = NULL; |
||||
} |
||||
if (inv->ncd_handle) { node_conn_direct_close(inv->ncd_handle); inv->ncd_handle = NULL; } |
||||
if (inv->temp_nq) { |
||||
queue_remove_data(inv->mgr->group->nodes, &inv->temp_nq->ll); |
||||
topo_nodeq_free_group_fields(inv->mgr->instance->topo_groups, inv->temp_nq); |
||||
queue_entry_free(&inv->temp_nq->ll); inv->temp_nq = NULL; |
||||
} |
||||
{ struct cm_invite_pending** pp = &inv->mgr->invite_list; |
||||
while (*pp) { if (*pp == inv) { *pp = inv->next; break; } pp = &(*pp)->next; } |
||||
} |
||||
if (inv->cb) inv->cb(inv->handle, inv->node_id, inv->mgr->group->group_id, CONN_EVENT_TIMEOUT, inv->cb_arg); |
||||
if (inv->handle) u_free(inv->handle); |
||||
cm_invite_cleanup(inv); |
||||
} |
||||
|
||||
void cm_invite_cleanup(struct cm_invite_pending* inv) { if (inv) u_free(inv); } |
||||
|
||||
static void cm_tcp_ready_cb(struct stcp_link* link, void* arg) { |
||||
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg; |
||||
if (!inv || inv->tcp_ready) return; inv->tcp_ready = 1; |
||||
struct ETCP_CONN* etcp = stcp_link_get_etcp_conn(link); etcp->peer_node_id = inv->node_id; |
||||
struct ll_entry* qe = queue_entry_new(sizeof(struct tcp_conn_entry)); |
||||
if (!qe) { cm_invite_fail(inv); return; } |
||||
{ struct tcp_conn_entry* te = (struct tcp_conn_entry*)qe->data; te->node_id=inv->node_id; te->link=link; te->etcp_conn=etcp; } |
||||
queue_data_put_with_index(inv->mgr->instance->tcp_connections, qe); |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: TCP invite ready node=0x%016llx", (unsigned long long)inv->node_id); |
||||
inv->state = CM_INVITE_WAIT_INFO; |
||||
{ struct CM_INVITE_REQ req; memset(&req,0,sizeof(req)); |
||||
req.cmd=ETCP_RT_ID_CONN_MGR; req.subcmd=CM_SUBCMD_INVITE_INFO_REQ; req.group_id=inv->mgr->group->group_id; |
||||
struct ll_entry* qe2 = queue_entry_new(0); if (!qe2) { cm_invite_fail(inv); return; } |
||||
qe2->dgram=u_malloc(sizeof(req)); if (!qe2->dgram) { queue_entry_free(qe2); cm_invite_fail(inv); return; } |
||||
memcpy(qe2->dgram,&req,sizeof(req)); qe2->len=sizeof(req); etcp_send(etcp, qe2); |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: TCP invite sent INVITE_INFO_REQ to 0x%016llx group=0x%016llx", |
||||
(unsigned long long)inv->node_id, (unsigned long long)inv->mgr->group->group_id); |
||||
} |
||||
} |
||||
|
||||
static void cm_tcp_close_cb(struct stcp_link* link, int err, void* arg) { |
||||
struct cm_invite_pending* inv = (struct cm_invite_pending*)arg; |
||||
if (!inv) return; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: TCP invite closed node=0x%016llx err=%d", (unsigned long long)inv->node_id, err); |
||||
inv->tcp_link = NULL; |
||||
} |
||||
|
||||
static int cm_invite_tcp_connect(struct cm_invite_pending* inv, const struct TOPO_NODE* ni) { |
||||
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) { |
||||
if (!(a->protocol & TOPO_PROTO_TCP)) continue; |
||||
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; memcpy(&sa,&sin,sizeof(sin)); |
||||
struct stcp_link_config cfg = {.ua=inv->mgr->instance->ua, .my_keys=&inv->mgr->instance->my_keys, .inst=inv->mgr->instance, |
||||
.peer_pubkey=ni->public_key, .peer_pubkey_mode=0, .remote_addr=&sa, .remote_port=a->port}; |
||||
inv->tcp_link = stcp_link_connect(&cfg); |
||||
if (inv->tcp_link) { stcp_link_set_on_ready(inv->tcp_link, cm_tcp_ready_cb, inv); stcp_link_set_on_close(inv->tcp_link, cm_tcp_close_cb, inv); return 1; } |
||||
return 0; |
||||
} |
||||
for (const struct TOPO_ADDR6* a6 = ni->v6_addrs; a6; a6 = a6->next) { |
||||
if (!(a6->protocol & TOPO_PROTO_TCP)) continue; |
||||
struct sockaddr_in6 sin6; memset(&sin6,0,sizeof(sin6)); sin6.sin6_family=AF_INET6; |
||||
memcpy(&sin6.sin6_addr,a6->addr,16); sin6.sin6_port=htons(a6->port); |
||||
struct sockaddr_storage sa; memcpy(&sa,&sin6,sizeof(sin6)); |
||||
struct stcp_link_config cfg = {.ua=inv->mgr->instance->ua, .my_keys=&inv->mgr->instance->my_keys, .inst=inv->mgr->instance, |
||||
.peer_pubkey=ni->public_key, .peer_pubkey_mode=0, .remote_addr=&sa, .remote_port=a6->port}; |
||||
inv->tcp_link = stcp_link_connect(&cfg); |
||||
if (inv->tcp_link) { stcp_link_set_on_ready(inv->tcp_link, cm_tcp_ready_cb, inv); stcp_link_set_on_close(inv->tcp_link, cm_tcp_close_cb, inv); return 1; } |
||||
return 0; |
||||
} |
||||
return 0; |
||||
} |
||||
|
||||
int conn_mgr_open_invite(struct UTUN_INSTANCE* inst, |
||||
uint64_t group_id, |
||||
struct TOPO_NODE* ni, |
||||
uint64_t node_id, |
||||
conn_mgr_cb_t cb, void* cb_arg, |
||||
struct CONN_MGR_HANDLE** out_handle) { |
||||
if (out_handle) *out_handle = NULL; |
||||
if (!inst || !inst->topo_groups || !cb) return -1; |
||||
|
||||
/* 1. Найти или создать группу → mgr */ |
||||
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id); |
||||
if (!group) { |
||||
char ch[32]; snprintf(ch, sizeof(ch), "%llu", (unsigned long long)group_id); |
||||
group = topo_groups_create_group(inst->topo_groups, group_id, TOPO_GROUP_TYPE_CHAT, ch); |
||||
if (!group) return -1; |
||||
} |
||||
struct CONN_MGR* mgr = group->conn_mgr; |
||||
if (!mgr) return -1; |
||||
|
||||
/* 2. ni == NULL → обычное 3-фазное подключение */ |
||||
if (!ni) return conn_mgr_open(mgr, node_id, 0, cb, cb_arg, out_handle); |
||||
|
||||
/* 3. ni != NULL → invite-режим */ |
||||
uint64_t nid = ni->node_id; |
||||
if (!nid || nid == inst->node_id) return -1; |
||||
if (!ni->v4_addrs && !ni->v6_addrs) return -1; |
||||
if (!ni->public_key[0]) return -1; |
||||
|
||||
/* 4. Копировать ni в group_ni */ |
||||
struct TOPO_GROUPS* groups = inst->topo_groups; |
||||
struct TOPO_NODE* gni = u_calloc(1, sizeof(*gni)); if (!gni) return -1; |
||||
gni->node_id=nid; gni->ver=1; gni->group_id=group_id; memcpy(gni->public_key,ni->public_key,SC_PUBKEY_SIZE); |
||||
for (const struct TOPO_ADDR4* src = ni->v4_addrs; src; src = src->next) { |
||||
struct TOPO_ADDR4* a4 = memory_pool_alloc(groups->v4_addr_pool); if (!a4) continue; |
||||
memset(a4,0,sizeof(*a4)); memcpy(a4->addr,src->addr,4); a4->port=src->port; |
||||
a4->type=TOPO_ADDR_NAT; a4->protocol=src->protocol; a4->next=gni->v4_addrs; gni->v4_addrs=a4; |
||||
} |
||||
for (const struct TOPO_ADDR6* src6 = ni->v6_addrs; src6; src6 = src6->next) { |
||||
struct TOPO_ADDR6* a6 = memory_pool_alloc(groups->v6_addr_pool); if (!a6) continue; |
||||
memset(a6,0,sizeof(*a6)); memcpy(a6->addr,src6->addr,16); a6->port=src6->port; |
||||
a6->type=TOPO_ADDR_NAT; a6->protocol=src6->protocol; a6->next=gni->v6_addrs; gni->v6_addrs=a6; |
||||
} |
||||
if (!gni->v4_addrs && !gni->v6_addrs) { u_free(gni); return -1; } |
||||
gni = topo_node_registry_store(groups, gni); if (!gni) return -1; |
||||
|
||||
/* 5. TOPO_GROUP_NODE + invite_pending */ |
||||
struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_GROUP_NODE)); |
||||
if (!qe) { topo_node_registry_unref(groups,gni->node_id); return -1; } |
||||
struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)qe; |
||||
memset((uint8_t*)nq+sizeof(struct ll_entry),0,sizeof(*nq)-sizeof(struct ll_entry)); nq->node_id=nid; |
||||
|
||||
struct cm_invite_pending* inv = u_calloc(1, sizeof(*inv)); |
||||
if (!inv) { queue_entry_free(qe); topo_node_registry_unref(groups,gni->node_id); return -1; } |
||||
inv->mgr=mgr; inv->node_id=nid; inv->state=CM_INVITE_CONNECTING; |
||||
inv->temp_nq=nq; inv->next=mgr->invite_list; mgr->invite_list=inv; |
||||
inv->cb=cb; inv->cb_arg=cb_arg; |
||||
|
||||
struct CONN_MGR_HANDLE* h = cm_invite_handle_new(NULL, NULL, nid, group_id); |
||||
if (!h) { cm_invite_fail(inv); return -1; } inv->handle=h; if (out_handle) *out_handle=h; |
||||
queue_data_put_with_index(group->nodes, &nq->ll); |
||||
|
||||
/* 6. NCD + TCP */ |
||||
int r = node_conn_direct_open_node(inst, nid, cm_invite_ncd_callback, inv, &inv->ncd_handle, gni, NULL); |
||||
if (r == NCD_ERR) { cm_invite_fail(inv); if (out_handle) *out_handle=NULL; return -1; } |
||||
int udp = (r == NCD_NEW || r == NCD_REUSED), tcp = cm_invite_tcp_connect(inv, gni); |
||||
if (!udp && !tcp) { cm_invite_fail(inv); if (out_handle) *out_handle=NULL; return -1; } |
||||
|
||||
inv->overall_timer = uasync_set_timeout(inst->ua, CM_INVITE_DEFAULT_TIMEOUT_MS*10, inv, cm_invite_overall_timeout, "conn_mgr_invite"); |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: invite started node=0x%016llx group=0x%016llx", |
||||
(unsigned long long)nid, (unsigned long long)group_id); |
||||
return 0; |
||||
} |
||||
@ -0,0 +1,145 @@
|
||||
/* conn_mgr_indirect.c — INDIRECT фаза + exchange-обработчики */ |
||||
#include "conn_mgr_priv.h" |
||||
#include "etcp.h" |
||||
#include <string.h> |
||||
#include "../lib/debug_config.h" |
||||
#include "../lib/mem.h" |
||||
#include "../lib/u_async.h" |
||||
#include "etcp_router.h" |
||||
#include "utun_instance.h" |
||||
#include "topo_group.h" |
||||
#include "route_connectivity.h" |
||||
|
||||
/* INDIRECT фаза: отправляет EXCHANGE_REQ с нашими лучшими кандидатами (best_candidates)
|
||||
* цели через BGP-маршрут. Ожидаем EXCHANGE_RESP — по RTT-замерам с обеих сторон |
||||
* выберем общего посредника с минимальной суммой RTT (cm_compute_intermediaries). */ |
||||
void cm_start_phase_indirect(struct CONN_MGR_ENTRY* entry) { |
||||
struct CONN_MGR* mgr = entry->mgr; |
||||
if (!mgr->best_candidate_count) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: phase 3 (indirect) no candidates for 0x%016llx", (unsigned long long)entry->node_id); |
||||
cm_deliver_event(entry, CONN_EVENT_TIMEOUT); return; |
||||
} |
||||
entry->main.phase = 3; entry->main.request_id = ++mgr->next_request_id; |
||||
struct CM_EXCHANGE_REQ req; memset(&req,0,sizeof(req)); |
||||
req.cmd=ETCP_RT_ID_CONN_MGR; req.subcmd=CM_SUBCMD_INTERM_EXCHANGE_REQ; req.request_id=entry->main.request_id; |
||||
req.candidate_count = mgr->best_candidate_count > 4 ? 4 : mgr->best_candidate_count; |
||||
for (uint8_t i=0;i<req.candidate_count;i++) req.candidates[i]=mgr->best_candidates[i]; |
||||
struct ll_entry* qe = queue_entry_new(0); if (!qe) return; |
||||
qe->dgram=u_malloc(sizeof(req)); memcpy(qe->dgram,&req,sizeof(req)); qe->len=sizeof(req); |
||||
etcp_route_send(mgr->instance, 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"); |
||||
struct cm_exchange_pending* ep = u_calloc(1,sizeof(*ep)); |
||||
if (ep) { ep->request_id=entry->main.request_id; ep->entry=entry; ep->timer=entry->main.timer; ep->next=mgr->exchange_pending; mgr->exchange_pending=ep; } |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: phase 3 sent EXCHANGE_REQ to 0x%016llx with %u candidates", |
||||
(unsigned long long)entry->node_id, req.candidate_count); |
||||
} |
||||
|
||||
void cm_exchange_timeout_cb(void* arg) { |
||||
struct CONN_MGR_ENTRY* entry = (struct CONN_MGR_ENTRY*)arg; entry->main.timer=NULL; |
||||
{ struct cm_exchange_pending** pp=&entry->mgr->exchange_pending; |
||||
while(*pp){if((*pp)->entry==entry){struct cm_exchange_pending* ep=*pp; *pp=ep->next; u_free(ep); break;} pp=&(*pp)->next;} } |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: exchange timeout for 0x%016llx", (unsigned long long)entry->node_id); |
||||
cm_deliver_event(entry, CONN_EVENT_TIMEOUT); |
||||
} |
||||
|
||||
/* Вычисляет лучших общих посредников по данным из EXCHANGE_RESP:
|
||||
* your_candidates — кандидаты инициатора + НАШ RTT до них |
||||
* my_candidates — НАШИ кандидаты + RTT цели до них |
||||
* Суммирует RTT для каждого общего узла, сортирует, выбирает топ-3. |
||||
* Отправляет INTERM_SELECTED инициатору и доставляет CONN_EVENT_UP. */ |
||||
void cm_compute_intermediaries(struct CONN_MGR_ENTRY* entry, struct CM_EXCHANGE_RESP* resp) { |
||||
struct { uint64_t nid; uint64_t rtt; } all[8]; uint8_t cnt=0; |
||||
for (uint8_t i=0;i<resp->your_count&&cnt<8;i++) { |
||||
uint16_t our=0; for(uint8_t j=0;j<CONN_MGR_MAX_CANDIDATES;j++) if(entry->mgr->best_candidates[j].node_id==resp->your_candidates[i].node_id){our=entry->mgr->best_candidates[j].rtt;break;} |
||||
all[cnt]=(typeof(all[0])){resp->your_candidates[i].node_id,(uint64_t)our+resp->your_candidates[i].rtt}; cnt++; |
||||
} |
||||
for (uint8_t i=0;i<resp->my_count&&cnt<8;i++) { |
||||
struct TOPO_GROUP_NODE* nq=topo_node_find_by_id(entry->mgr->group,resp->my_candidates[i].node_id); |
||||
uint16_t our=nq?cm_get_node_min_rtt(nq):0; if(our==0xFFFF)our=0; |
||||
all[cnt]=(typeof(all[0])){resp->my_candidates[i].node_id,(uint64_t)our+resp->my_candidates[i].rtt}; cnt++; |
||||
} |
||||
for (uint8_t i=0;i<cnt;i++) for(uint8_t j=i+1;j<cnt;j++) if(all[j].rtt<all[i].rtt){typeof(all[0])t=all[i];all[i]=all[j];all[j]=t;} |
||||
uint8_t sel=cnt>CONN_MGR_MAX_INTERMEDIARIES?CONN_MGR_MAX_INTERMEDIARIES:cnt; |
||||
for(uint8_t i=0;i<sel;i++)entry->intermediaries[i]=all[i].nid; |
||||
entry->intermediariy_count=sel; entry->rr_idx=0; |
||||
struct CM_INTERM_SEL pkt; memset(&pkt,0,sizeof(pkt)); |
||||
pkt.cmd=ETCP_RT_ID_CONN_MGR; pkt.subcmd=CM_SUBCMD_INTERM_SELECTED; pkt.count=sel; |
||||
for(uint8_t i=0;i<sel;i++){pkt.selected[i].node_id=all[i].nid;pkt.selected[i].rtt=0; |
||||
struct TOPO_GROUP_NODE* nq=topo_node_find_by_id(entry->mgr->group,all[i].nid); 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,entry->mgr->group->group_id,entry->node_id,qe,1);} |
||||
entry->conn_type=CONN_TYPE_INDIRECT; entry->state=CONN_MGR_STATE_CONNECTED; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL,"conn_mgr: indirect OK for 0x%016llx via %u intermediaries",(unsigned long long)entry->node_id,sel); |
||||
cm_update_nodeinfo(entry); cm_deliver_event(entry,CONN_EVENT_UP); |
||||
} |
||||
|
||||
void cm_exchange_probe_retry_cb(void* arg) { |
||||
struct cm_exchange_pending* ep=(struct cm_exchange_pending*)arg; ep->timer=NULL; |
||||
uint8_t need=0; uint64_t now=get_time_tb(); |
||||
for(uint8_t i=0;i<ep->cached.my_count&&i<4;i++){ |
||||
struct TOPO_GROUP_NODE* nq=topo_node_find_by_id(ep->entry->mgr->group,ep->cached.my_candidates[i].node_id); |
||||
if(!nq||!cm_is_rtt_fresh(nq,now)){need++;break;} |
||||
} |
||||
if(!need){cm_compute_intermediaries(ep->entry,&ep->cached);struct cm_exchange_pending**pp=&ep->entry->mgr->exchange_pending;while(*pp){if(*pp==ep){*pp=ep->next;break;}pp=&(*pp)->next;}u_free(ep);} |
||||
else ep->timer=uasync_set_timeout(ep->entry->mgr->instance->ua,2000,ep,cm_exchange_probe_retry_cb,"cm_exch_probe"); |
||||
} |
||||
|
||||
/* Инициатор получил EXCHANGE_RESP: кеширует ответ, проверяет свежесть RTT
|
||||
* до СВОИХ кандидатов (тех, что в my_candidates ответа). Если протухли — |
||||
* probe + retry-таймер 2с. Если свежие — cm_compute_intermediaries. */ |
||||
void cm_handle_interm_exchange_resp(struct CONN_MGR* mgr, const uint8_t* data, size_t len) { |
||||
const struct CM_EXCHANGE_RESP* resp=(const struct CM_EXCHANGE_RESP*)data; |
||||
if(len<offsetof(struct CM_EXCHANGE_RESP,your_candidates))return; |
||||
struct cm_exchange_pending* ep=mgr->exchange_pending; |
||||
while(ep){if(ep->request_id==resp->request_id)break;ep=ep->next;} |
||||
if(!ep)return; |
||||
if(ep->entry->main.timer){uasync_cancel_timeout(mgr->instance->ua,ep->entry->main.timer);ep->entry->main.timer=NULL;} |
||||
memcpy(&ep->cached,resp,len<sizeof(ep->cached)?len:sizeof(ep->cached)); ep->received=1; ep->probed=0; |
||||
uint8_t need=0; uint64_t now=get_time_tb(); |
||||
for(uint8_t i=0;i<ep->cached.my_count&&i<4;i++){ |
||||
struct TOPO_GROUP_NODE* nq=topo_node_find_by_id(mgr->group,ep->cached.my_candidates[i].node_id); |
||||
if(!nq||!cm_is_rtt_fresh(nq,now)){need++;route_connectivity_probe_node(mgr->instance,nq);} |
||||
} |
||||
if(!need){ep->probed=1;cm_compute_intermediaries(ep->entry,&ep->cached);struct cm_exchange_pending**pp=&mgr->exchange_pending;while(*pp){if(*pp==ep){*pp=ep->next;break;}pp=&(*pp)->next;}u_free(ep);return;} |
||||
ep->timer=uasync_set_timeout(mgr->instance->ua,2000,ep,cm_exchange_probe_retry_cb,"cm_exch_probe"); |
||||
} |
||||
|
||||
/* Финальный выбор посредников от инициатора (приходит второй стороне INDIRECT-
|
||||
* обмена). Сохраняет выбранные intermediaries в entry, помечает CONNECTED. */ |
||||
void cm_handle_interm_selected(struct CONN_MGR* mgr, const uint8_t* data, size_t len) { |
||||
const struct CM_INTERM_SEL* sel=(const struct CM_INTERM_SEL*)data; |
||||
if(len<offsetof(struct CM_INTERM_SEL,selected))return; |
||||
struct CONN_MGR_ENTRY* entry=NULL; |
||||
for(size_t i=0;i<mgr->entry_count;i++) if(mgr->entries[i].state==CONN_MGR_STATE_CONNECTING){entry=&mgr->entries[i];break;} |
||||
if(!entry)return; |
||||
uint8_t c=sel->count>CONN_MGR_MAX_INTERMEDIARIES?CONN_MGR_MAX_INTERMEDIARIES:sel->count; |
||||
for(uint8_t i=0;i<c;i++)entry->intermediaries[i]=sel->selected[i].node_id; |
||||
entry->intermediariy_count=c; entry->rr_idx=0; entry->conn_type=CONN_TYPE_INDIRECT; entry->state=CONN_MGR_STATE_CONNECTED; |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL,"conn_mgr: got INTERM_SELECTED count=%u",c); |
||||
} |
||||
|
||||
/* Цель получила EXCHANGE_REQ: измеряет RTT до кандидатов инициатора (если
|
||||
* замеры протухли — запускает probe), отправляет EXCHANGE_RESP со СВОИМИ |
||||
* кандидатами + замерами RTT до кандидатов инициатора. */ |
||||
void cm_handle_interm_exchange_req(struct ETCP_CONN* conn, struct CM_EXCHANGE_REQ* req) { |
||||
struct CONN_MGR* mgr=conn->instance->conn_mgr; if(!mgr)return; |
||||
uint64_t now=get_time_tb(); |
||||
for(uint8_t i=0;i<req->candidate_count&&i<4;i++){ |
||||
struct TOPO_GROUP_NODE* nq=topo_node_find_by_id(mgr->group,req->candidates[i].node_id); |
||||
if(nq&&cm_is_rtt_fresh(nq,now))continue; route_connectivity_probe_node(mgr->instance,nq); |
||||
} |
||||
struct CM_EXCHANGE_RESP resp; memset(&resp,0,sizeof(resp)); |
||||
resp.cmd=ETCP_RT_ID_CONN_MGR; resp.subcmd=CM_SUBCMD_INTERM_EXCHANGE_RESP; resp.request_id=req->request_id; |
||||
resp.my_count=mgr->best_candidate_count>4?4:mgr->best_candidate_count; |
||||
for(uint8_t i=0;i<resp.my_count;i++)resp.my_candidates[i]=mgr->best_candidates[i]; |
||||
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_GROUP_NODE* nq2=topo_node_find_by_id(mgr->group,req->candidates[i].node_id); |
||||
resp.your_candidates[i].rtt=nq2?cm_get_node_min_rtt(nq2):0; |
||||
} |
||||
size_t sz=offsetof(struct CM_EXCHANGE_RESP,your_candidates)+(size_t)resp.your_count*sizeof(struct CONN_MGR_CANDIDATE); |
||||
struct ll_entry* qe=queue_entry_new(0); |
||||
if(qe){qe->dgram=u_malloc(sz);memcpy(qe->dgram,&resp,sz);qe->len=(uint16_t)sz;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); |
||||
} |
||||
@ -0,0 +1,174 @@
|
||||
/* conn_mgr_monitor.c — bg_ping, candidate_ping, idle, best_candidates */ |
||||
#include "conn_mgr_priv.h" |
||||
#include <string.h> |
||||
#include "../lib/platform_compat.h" |
||||
#include "../lib/debug_config.h" |
||||
#include "../lib/mem.h" |
||||
#include "../lib/u_async.h" |
||||
#include "utun_instance.h" |
||||
#include "etcp.h" |
||||
#include "etcp_connections.h" |
||||
#include "etcp_router.h" |
||||
#include "topo_node.h" |
||||
#include "topo_group.h" |
||||
#include "route_connectivity.h" |
||||
#include "route_ping.h" |
||||
|
||||
/* ═══════ RTT утилиты ═══════ */ |
||||
|
||||
uint16_t cm_get_node_min_rtt(struct TOPO_GROUP_NODE* nq) { |
||||
uint16_t r = 0xFFFF; |
||||
if (nq->connectivity.interface_status == PROBE_RESULT_REACHABLE && nq->connectivity.interface_min_rtt < r) r = nq->connectivity.interface_min_rtt; |
||||
if (nq->connectivity.nat_status == PROBE_RESULT_REACHABLE && nq->connectivity.nat_min_rtt < r) r = nq->connectivity.nat_min_rtt; |
||||
if (nq->connectivity.real_status == PROBE_RESULT_REACHABLE && nq->connectivity.real_min_rtt < r) r = nq->connectivity.real_min_rtt; |
||||
return r; |
||||
} |
||||
|
||||
uint64_t cm_get_node_max_probe_time(struct TOPO_GROUP_NODE* nq) { |
||||
uint64_t t = nq->connectivity.interface_probe_time; |
||||
if (nq->connectivity.nat_probe_time > t) t = nq->connectivity.nat_probe_time; |
||||
if (nq->connectivity.real_probe_time > t) t = nq->connectivity.real_probe_time; |
||||
return t; |
||||
} |
||||
|
||||
int cm_is_rtt_fresh(struct TOPO_GROUP_NODE* nq, uint64_t now_tb) { |
||||
return cm_get_node_max_probe_time(nq) > 0 && (now_tb - cm_get_node_max_probe_time(nq)) < (uint64_t)(CACHE_MS * 10); |
||||
} |
||||
|
||||
/* ═══════ best_candidates ═══════ */ |
||||
|
||||
/* Обновить RTT кандидата-посредника в отсортированном списке (≤3). Вызывается
|
||||
* из route_connectivity после успешного probe. INDIRECT-фаза использует этот |
||||
* список (best_candidates) для выбора посредника с минимальным суммарным RTT. */ |
||||
void conn_mgr_update_best_candidates(struct CONN_MGR* mgr, uint64_t node_id, uint16_t rtt) { |
||||
if (!mgr || node_id == mgr->instance->node_id || !rtt) return; |
||||
for (uint8_t i = 0; i < mgr->best_candidate_count; i++) { |
||||
if (mgr->best_candidates[i].node_id != node_id) continue; |
||||
mgr->best_candidates[i].rtt = rtt; |
||||
for (uint8_t j = i; j > 0 && mgr->best_candidates[j].rtt < mgr->best_candidates[j-1].rtt; j--) |
||||
{ struct CONN_MGR_CANDIDATE t = mgr->best_candidates[j]; mgr->best_candidates[j]=mgr->best_candidates[j-1]; mgr->best_candidates[j-1]=t; } |
||||
for (uint8_t j = i; j + 1 < mgr->best_candidate_count && mgr->best_candidates[j].rtt > mgr->best_candidates[j+1].rtt; j++) |
||||
{ struct CONN_MGR_CANDIDATE t = mgr->best_candidates[j]; mgr->best_candidates[j]=mgr->best_candidates[j+1]; mgr->best_candidates[j+1]=t; } |
||||
return; |
||||
} |
||||
if (mgr->best_candidate_count < CONN_MGR_MAX_CANDIDATES) |
||||
{ mgr->best_candidates[mgr->best_candidate_count++]=(struct CONN_MGR_CANDIDATE){node_id,rtt}; } |
||||
else if (rtt < mgr->best_candidates[CONN_MGR_MAX_CANDIDATES-1].rtt) |
||||
{ mgr->best_candidates[CONN_MGR_MAX_CANDIDATES-1]=(struct CONN_MGR_CANDIDATE){node_id,rtt}; } |
||||
else return; |
||||
for (uint8_t i = mgr->best_candidate_count - 1; i > 0 && mgr->best_candidates[i].rtt < mgr->best_candidates[i-1].rtt; i--) |
||||
{ struct CONN_MGR_CANDIDATE t = mgr->best_candidates[i]; mgr->best_candidates[i]=mgr->best_candidates[i-1]; mgr->best_candidates[i-1]=t; } |
||||
} |
||||
|
||||
/* ═══════ idle ═══════ */ |
||||
|
||||
/* Проверка неактивности: если трафика не было > idle_timeout_ms — доставляет
|
||||
* CONN_EVENT_TIMEOUT всем handle'ам и чистит entry. Таймер самоперезапускается. */ |
||||
void cm_idle_timer_cb(void* arg) { |
||||
struct CONN_MGR_ENTRY* entry = (struct CONN_MGR_ENTRY*)arg; |
||||
if (entry->state != CONN_MGR_STATE_CONNECTED || !entry->idle_timeout_ms) return; |
||||
uint64_t idle = get_time_tb() - entry->last_traffic_tb; |
||||
if (idle > (uint64_t)entry->idle_timeout_ms * 10) { |
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: node 0x%016llx idle %llums, disconnecting", |
||||
(unsigned long long)entry->node_id, (unsigned long long)(idle / 10)); |
||||
cm_deliver_event(entry, CONN_EVENT_TIMEOUT); cm_entry_cleanup(entry); return; |
||||
} |
||||
entry->idle_timer = uasync_set_timeout(entry->mgr->instance->ua, CONN_MGR_IDLE_CHECK_INTERVAL_TB, entry, cm_idle_timer_cb, "conn_mgr_idle"); |
||||
} |
||||
|
||||
/* ═══════ bg_ping ═══════ */ |
||||
|
||||
static void cm_bg_ping_noop_cb(int s, uint16_t r, void* a, uint64_t n, const uint8_t* d, size_t l) { (void)s;(void)r;(void)a;(void)n;(void)d;(void)l; } |
||||
|
||||
static void cm_bg_ping_to_node(struct CONN_MGR* mgr, struct TOPO_NODE* ni) { |
||||
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) { |
||||
if (a->type != TOPO_ADDR_NAT && a->type != TOPO_ADDR_INTERFACE) continue; |
||||
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; memcpy(&sa,&sin,sizeof(sin)); |
||||
struct ETCP_SOCKET* s = mgr->instance->etcp_sockets; |
||||
while (s) { if (s->local_addr.ss_family == AF_INET) break; s = s->next; } |
||||
if (!s) break; |
||||
etcp_send_ping_to_socket(mgr->instance, s, ni->public_key, &sa, CONN_PROBE_TIMEOUT_MS, cm_bg_ping_noop_cb, NULL, NULL, 0, 0); |
||||
return; |
||||
} |
||||
for (const struct TOPO_ADDR6* a6 = ni->v6_addrs; a6; a6 = a6->next) { |
||||
if (a6->type != TOPO_ADDR_NAT && a6->type != TOPO_ADDR_INTERFACE) continue; |
||||
struct sockaddr_in6 sin6; memset(&sin6,0,sizeof(sin6)); sin6.sin6_family=AF_INET6; |
||||
memcpy(&sin6.sin6_addr,a6->addr,16); sin6.sin6_port=htons(a6->port); |
||||
struct sockaddr_storage sa; memcpy(&sa,&sin6,sizeof(sin6)); |
||||
struct ETCP_SOCKET* s = mgr->instance->etcp_sockets; |
||||
while (s) { if (s->local_addr.ss_family == AF_INET6) break; s = s->next; } |
||||
if (!s) break; |
||||
etcp_send_ping_to_socket(mgr->instance, s, ni->public_key, &sa, CONN_PROBE_TIMEOUT_MS, cm_bg_ping_noop_cb, NULL, NULL, 0, 0); |
||||
return; |
||||
} |
||||
} |
||||
|
||||
/* Фоновый ping: один узел группы за тик (~100ms), полный цикл ≥ 10с.
|
||||
* Пингует только если нет активного ETCP-соединения с узлом. Цель: поддерживать |
||||
* свежие RTT-замеры для ВСЕХ узлов группы — без этого best_candidates протухают |
||||
* и INDIRECT-фаза не может выбрать посредника. */ |
||||
void cm_bg_ping_timer_cb(void* arg) { |
||||
struct CONN_MGR* mgr = (struct CONN_MGR*)arg; |
||||
struct TOPO_GROUP* group = mgr->group; |
||||
if (!group || !group->nodes) goto rearm; |
||||
size_t total = queue_entry_count(group->nodes); |
||||
if (!total) goto rearm; |
||||
|
||||
uint64_t now = get_time_tb(); |
||||
if (mgr->bg_ping_cursor >= total) { |
||||
mgr->bg_ping_cursor = 0; |
||||
if ((now - mgr->bg_ping_cycle_start_tb) < CONN_MGR_BG_PING_CYCLE_MIN_TB) |
||||
{ mgr->bg_ping_timer = uasync_set_timeout(mgr->instance->ua, (int)(CONN_MGR_BG_PING_CYCLE_MIN_TB - (now - mgr->bg_ping_cycle_start_tb)), mgr, cm_bg_ping_timer_cb, "conn_mgr_bg_ping"); return; } |
||||
mgr->bg_ping_cycle_start_tb = now; |
||||
} |
||||
|
||||
struct ll_entry* e = group->nodes->head; |
||||
for (size_t c = 0; e && c < mgr->bg_ping_cursor; c++) e = e->next; |
||||
if (e) { |
||||
struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e; |
||||
if (nq->node_id != mgr->instance->node_id) { |
||||
int has_active = 0; |
||||
struct ETCP_CONN* dc = topo_group_find_conn_for_node(group, nq->node_id); |
||||
if (dc && dc->links) { struct ETCP_LINK* l = dc->links; while (l) { if (l->initialized && l->link_status) { has_active = 1; break; } l = l->next; } } |
||||
if (!has_active) { |
||||
struct TOPO_NODE* bg_ni = topo_node_registry_find(mgr->instance->topo_groups, nq->node_id); |
||||
if (bg_ni) cm_bg_ping_to_node(mgr, bg_ni); |
||||
} |
||||
} |
||||
mgr->bg_ping_cursor++; |
||||
} |
||||
rearm: |
||||
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"); |
||||
} |
||||
|
||||
/* ═══════ candidate_ping ═══════ */ |
||||
|
||||
static int cm_has_active_conn(struct TOPO_GROUP* g, uint64_t nid) { |
||||
struct ETCP_CONN* c = topo_group_find_conn_for_node(g, nid); |
||||
if (!c || !c->links) return 0; |
||||
struct ETCP_LINK* l = c->links; while (l) { if (l->initialized && l->link_status) return 1; l = l->next; } |
||||
return 0; |
||||
} |
||||
|
||||
/* Обновление кандидатов-посредников: удаляет протухших (>30с без свежих RTT),
|
||||
* активных пингует напрямую, неактивных probe'ит через route_connectivity. |
||||
* Вызывается раз в ~2с, самоперезапускается. */ |
||||
void cm_candidate_ping_timer_cb(void* arg) { |
||||
struct CONN_MGR* mgr = (struct CONN_MGR*)arg; uint64_t now = get_time_tb(); |
||||
for (uint8_t i = 0; i < mgr->best_candidate_count; i++) { |
||||
uint64_t nid = mgr->best_candidates[i].node_id; |
||||
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(mgr->group, nid); |
||||
if (!nq) { memmove(&mgr->best_candidates[i], &mgr->best_candidates[i+1], (--mgr->best_candidate_count - i) * sizeof(mgr->best_candidates[0])); i--; continue; } |
||||
if (cm_get_node_max_probe_time(nq) > 0 && (now - cm_get_node_max_probe_time(nq)) >= CONN_MGR_CANDIDATE_STALE_TB) |
||||
{ DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: candidate 0x%016llx stale, removing", (unsigned long long)nid); |
||||
memmove(&mgr->best_candidates[i], &mgr->best_candidates[i+1], (--mgr->best_candidate_count - i) * sizeof(mgr->best_candidates[0])); i--; continue; } |
||||
if (cm_has_active_conn(mgr->group, nid)) { |
||||
struct ETCP_CONN* dc = topo_group_find_conn_for_node(mgr->group, nid); |
||||
struct TOPO_NODE* ni = topo_node_registry_find(mgr->instance->topo_groups, nq->node_id); |
||||
if (ni && dc) etcp_send_ping(mgr->instance, ni->public_key, &dc->links->remote_addr, CONN_PROBE_TIMEOUT_MS, NULL, NULL, NULL, 0); |
||||
} else route_connectivity_probe_node(mgr->instance, nq); |
||||
} |
||||
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"); |
||||
} |
||||
@ -0,0 +1,197 @@
|
||||
#ifndef CONN_MGR_PRIV_H |
||||
#define CONN_MGR_PRIV_H |
||||
|
||||
#include "conn_mgr.h" |
||||
#include <stdint.h> |
||||
#include "../lib/ll_queue.h" |
||||
#include "etcp_api.h" |
||||
#include "secure_channel.h" |
||||
#include "../transport_layer/node_conn_direct.h" |
||||
#include "topo_node.h" |
||||
|
||||
struct UTUN_INSTANCE; |
||||
struct ETCP_CONN; |
||||
struct ETCP_LINK; |
||||
struct ETCP_SOCKET; |
||||
struct ETCP_ROUTER_CONN; |
||||
struct TOPO_GROUP_NODE; |
||||
struct TOPO_GROUP; |
||||
struct stcp_link; |
||||
|
||||
/* ═══════ константы ═══════ */ |
||||
|
||||
#define CONN_MGR_MAX_CANDIDATES 3 |
||||
#define CONN_MGR_CANDIDATE_STALE_TB 300000 |
||||
#define CONN_MGR_CANDIDATE_PING_TB 20000 |
||||
#define CONN_MGR_BG_PING_INTERVAL_TB 1000 |
||||
#define CONN_MGR_BG_PING_CYCLE_MIN_TB 100000 |
||||
#define CONN_MGR_IDLE_CHECK_INTERVAL_TB 10000 |
||||
#define CONN_MGR_LOCAL_SCAN_ATTEMPTS 3 |
||||
#define CONN_MGR_LOCAL_SCAN_TIMEOUT_MS 100 |
||||
#define CONN_MGR_CONNECT_DIRECT_TIMEOUT_MS 2000 |
||||
#define CONN_MGR_CONNECT_REVERSE_TIMEOUT_MS 15000 |
||||
#define CONN_MGR_INTERM_EXCHANGE_TIMEOUT_MS 15000 |
||||
#define CM_INVITE_DEFAULT_TIMEOUT_MS 2000 |
||||
#define CACHE_MS 20000 /* время жизни RTT-замера (через cm_is_rtt_fresh) */ |
||||
#define CONN_MGR_MAX_INTERMEDIARIES 3 |
||||
|
||||
/* ═══════ состояния ═══════ */ |
||||
|
||||
enum CONN_MGR_STATE { |
||||
CONN_MGR_STATE_DISCONNECTED = 0, |
||||
CONN_MGR_STATE_CONNECTING = 1, |
||||
CONN_MGR_STATE_CONNECTED = 2, |
||||
}; |
||||
|
||||
enum CM_TRY_STATE { |
||||
CM_TRY_NONE = 0, |
||||
CM_TRY_PENDING = 1, |
||||
CM_TRY_OK = 2, |
||||
CM_TRY_FAILED = 3, |
||||
}; |
||||
|
||||
/* ═══════ протокол CONN_MGR (ETCP_RT_ID_CONN_MGR) ═══════ */ |
||||
|
||||
enum { |
||||
CM_SUBCMD_DIRECT_REQ = 0x01, |
||||
CM_SUBCMD_DIRECT_RESP = 0x02, |
||||
CM_SUBCMD_INTERM_EXCHANGE_REQ = 0x03, |
||||
CM_SUBCMD_INTERM_EXCHANGE_RESP= 0x04, |
||||
CM_SUBCMD_INTERM_SELECTED = 0x05, |
||||
CM_SUBCMD_DISCONNECT = 0x06, |
||||
CM_SUBCMD_BRIDGE_REQUEST = 0x10, |
||||
CM_SUBCMD_BRIDGE_RESPONSE = 0x11, |
||||
CM_SUBCMD_BRIDGE_CLOSE = 0x12, |
||||
CM_SUBCMD_BRIDGE_CLOSED = 0x13, |
||||
CM_SUBCMD_INVITE_INFO_REQ = 0x0A, |
||||
CM_SUBCMD_INVITE_INFO_RESP = 0x0B, |
||||
}; |
||||
|
||||
struct CONN_MGR_CANDIDATE { uint64_t node_id; uint16_t rtt; } __attribute__((packed)); |
||||
|
||||
#pragma pack(push, 1) |
||||
struct CM_DIRECT_REQ { uint8_t cmd, subcmd; uint32_t request_id; uint8_t addr_count; }; |
||||
struct CM_DIRECT_RESP { uint8_t cmd, subcmd; uint32_t request_id; uint8_t accepted; }; |
||||
struct CM_EXCHANGE_REQ { uint8_t cmd, subcmd; uint32_t request_id; uint8_t candidate_count; struct CONN_MGR_CANDIDATE candidates[4]; }; |
||||
struct CM_EXCHANGE_RESP { uint8_t cmd, subcmd; uint32_t request_id; uint8_t my_count, your_count; struct CONN_MGR_CANDIDATE my_candidates[4], your_candidates[4]; }; |
||||
struct CM_INTERM_SEL { uint8_t cmd, subcmd; uint32_t request_id; uint8_t count; struct CONN_MGR_CANDIDATE selected[3]; }; |
||||
struct CM_DISCONNECT { uint8_t cmd, subcmd; uint64_t node_id; }; |
||||
#pragma pack(pop) |
||||
|
||||
#pragma pack(push, 1) |
||||
struct CM_INVITE_REQ { uint8_t cmd, subcmd; uint64_t group_id; }; |
||||
#pragma pack(pop) |
||||
|
||||
#pragma pack(push, 1) |
||||
struct CM_INVITE_RESP { uint8_t cmd, subcmd; uint64_t group_id; uint8_t is_member; uint8_t reserved[3]; uint8_t ed25519_pubkey[SC_PUBKEY_SIZE]; uint8_t node_name_len; uint8_t node_name[0]; }; |
||||
#pragma pack(pop) |
||||
|
||||
#define CM_DIRECT_REQ_H_SIZE (sizeof(struct CM_DIRECT_REQ)) |
||||
#define CM_DIRECT_RESP_HDR_SIZE (sizeof(struct CM_DIRECT_RESP)) |
||||
#define CM_EXCHANGE_REQ_SIZE (sizeof(struct CM_EXCHANGE_REQ)) |
||||
#define CM_EXCHANGE_RESP_SIZE (sizeof(struct CM_EXCHANGE_RESP)) |
||||
#define CM_INTERM_SEL_SIZE (sizeof(struct CM_INTERM_SEL)) |
||||
#define CM_DISCONNECT_SIZE (sizeof(struct CM_DISCONNECT)) |
||||
#define CM_INVITE_REQ_SIZE (sizeof(struct CM_INVITE_REQ)) |
||||
|
||||
/* ═══════ внутренние структуры ═══════ */ |
||||
|
||||
struct cm_reverse_pending { struct cm_reverse_pending* next; uint32_t request_id; struct CONN_MGR_ENTRY* entry; }; |
||||
|
||||
struct cm_exchange_pending { |
||||
struct cm_exchange_pending* next; uint32_t request_id; |
||||
struct CONN_MGR_ENTRY* entry; void* timer; |
||||
struct CM_EXCHANGE_RESP cached; uint8_t received, probed; |
||||
}; |
||||
|
||||
enum CM_INVITE_STATE { CM_INVITE_CONNECTING = 0, CM_INVITE_WAIT_INFO = 1, }; |
||||
|
||||
struct cm_invite_pending { |
||||
struct cm_invite_pending* next; struct CONN_MGR* mgr; |
||||
uint64_t node_id; uint8_t state; |
||||
struct NODE_CONN_DIRECT* ncd_handle; struct stcp_link* tcp_link; |
||||
uint8_t tcp_ready:1; void* overall_timer; |
||||
struct CONN_MGR_HANDLE* handle; struct TOPO_GROUP_NODE* temp_nq; |
||||
conn_mgr_cb_t cb; void* cb_arg; |
||||
}; |
||||
|
||||
struct CONN_MGR_HANDLE { |
||||
uint64_t node_id; uint64_t group_id; conn_mgr_cb_t cb; void* cb_arg; |
||||
struct CONN_MGR_ENTRY* entry; struct CONN_MGR_HANDLE* next; |
||||
}; |
||||
|
||||
struct CONN_MGR_ENTRY { |
||||
uint64_t node_id; uint8_t state, conn_type; |
||||
uint32_t idle_timeout_ms; uint64_t last_traffic_tb; void* idle_timer; |
||||
struct CONN_MGR_HANDLE* handles; struct NODE_CONN_DIRECT* ncd_handle; |
||||
uint64_t intermediaries[3]; uint8_t intermediariy_count, rr_idx; |
||||
uint8_t local_scan_state, main_connect_state, db_loaded; |
||||
struct { uint8_t phase; void* timer; uint32_t request_id; void* conn_ctx; } main; |
||||
struct CONN_MGR* mgr; |
||||
}; |
||||
|
||||
struct CONN_MGR { |
||||
struct UTUN_INSTANCE* instance; struct TOPO_GROUP* group; |
||||
struct CONN_MGR_ENTRY* entries; size_t entry_count, entry_capacity; |
||||
void *bg_ping_timer, *candidate_ping_timer; |
||||
size_t bg_ping_cursor; uint64_t bg_ping_cycle_start_tb; |
||||
struct CONN_MGR_CANDIDATE best_candidates[CONN_MGR_MAX_CANDIDATES]; uint8_t best_candidate_count; |
||||
uint32_t next_request_id, direct_timeout_ms; uint8_t initialized; |
||||
struct cm_reverse_pending* reverse_pending; |
||||
struct cm_exchange_pending* exchange_pending; |
||||
struct cm_invite_pending* invite_list; |
||||
uint8_t _unused[0]; |
||||
}; |
||||
|
||||
/* ═══════ декларации внутренних функций ═══════ */ |
||||
|
||||
/* core — публичные (в priv.h, не в conn_mgr.h) */ |
||||
int conn_mgr_open(struct CONN_MGR* mgr, uint64_t node_id, uint32_t idle_timeout_ms, |
||||
conn_mgr_cb_t cb, void* cb_arg, struct CONN_MGR_HANDLE** out_handle); |
||||
|
||||
/* core */ |
||||
struct CONN_MGR_ENTRY* cm_find_entry(struct CONN_MGR* mgr, uint64_t node_id); |
||||
struct CONN_MGR_ENTRY* cm_ensure_entry(struct CONN_MGR* mgr, uint64_t node_id); |
||||
void cm_entry_cleanup(struct CONN_MGR_ENTRY* entry); |
||||
void cm_cleanup_db_node(struct CONN_MGR_ENTRY* entry); |
||||
void cm_deliver_event(struct CONN_MGR_ENTRY* entry, enum conn_mgr_event event); |
||||
void cm_update_nodeinfo(struct CONN_MGR_ENTRY* entry); |
||||
void cm_clear_nodeinfo(struct CONN_MGR* mgr, uint64_t node_id); |
||||
void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void* arg); |
||||
void cm_invite_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ncd_ev, void* arg); |
||||
void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry); |
||||
void cm_start_phase_reverse(struct CONN_MGR_ENTRY* entry); |
||||
void cm_start_phase_indirect(struct CONN_MGR_ENTRY* entry); |
||||
void cm_start_local_scan(struct CONN_MGR_ENTRY* entry); |
||||
void cm_handle_direct_req(struct ETCP_CONN* conn, const uint8_t* data, size_t len); |
||||
void cm_handle_disconnect(struct CONN_MGR* mgr, uint64_t node_id); |
||||
void cm_reverse_init_cb(struct ETCP_CONN* conn, int event, void* arg); |
||||
void cm_reverse_timeout_cb(void* arg); |
||||
void cm_invite_fail(struct cm_invite_pending* inv); |
||||
void cm_invite_cleanup(struct cm_invite_pending* inv); |
||||
int cm_nat_compatible(struct ETCP_SOCKET* our, uint8_t tcfg, uint8_t tnat); |
||||
void cm_add_v4_link(struct ETCP_CONN* conn, const uint8_t* addr, uint16_t port, struct ETCP_SOCKET* s); |
||||
void cm_add_v6_link(struct ETCP_CONN* conn, const uint8_t addr[16], uint16_t port, struct ETCP_SOCKET* s); |
||||
int cm_has_direct_ip(struct CONN_MGR* mgr, struct TOPO_GROUP_NODE* nq); |
||||
int cm_has_local_addr(struct CONN_MGR* mgr, struct TOPO_GROUP_NODE* nq); |
||||
uint8_t cm_classify_v6_addr(const uint8_t addr[16]); |
||||
uint8_t cm_sock_v6_classify(const struct ETCP_SOCKET* s); |
||||
|
||||
/* indirect */ |
||||
void cm_exchange_timeout_cb(void* arg); |
||||
void cm_compute_intermediaries(struct CONN_MGR_ENTRY* entry, struct CM_EXCHANGE_RESP* resp); |
||||
void cm_exchange_probe_retry_cb(void* arg); |
||||
void cm_handle_interm_exchange_req(struct ETCP_CONN* conn, struct CM_EXCHANGE_REQ* req); |
||||
void cm_handle_interm_exchange_resp(struct CONN_MGR* mgr, const uint8_t* data, size_t len); |
||||
void cm_handle_interm_selected(struct CONN_MGR* mgr, const uint8_t* data, size_t len); |
||||
|
||||
/* monitor */ |
||||
void cm_idle_timer_cb(void* arg); |
||||
void cm_bg_ping_timer_cb(void* arg); |
||||
void cm_candidate_ping_timer_cb(void* arg); |
||||
uint16_t cm_get_node_min_rtt(struct TOPO_GROUP_NODE* nq); |
||||
uint64_t cm_get_node_max_probe_time(struct TOPO_GROUP_NODE* nq); |
||||
int cm_is_rtt_fresh(struct TOPO_GROUP_NODE* nq, uint64_t now_tb); |
||||
void cm_invite_overall_timeout(void* arg); |
||||
|
||||
#endif |
||||
Loading…
Reference in new issue