utun2/src/routing_layer/conn_mgr_monitor.c

/* 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"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif

/* ═══════ probe time утилиты ═══════ */

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; }
}

/* Маршрут может вести через посредника. Для прямой пробы нужны transport peer и его живой линк. */
static struct ETCP_LINK* cm_active_direct_link(struct TOPO_GROUP* group, uint64_t node_id) {
    struct ETCP_CONN* conn = instance_find_conn(group->instance,node_id);
    if (!conn || conn->state != 1 || conn->close_requested) return NULL;
    for (struct ETCP_LINK* link = conn->links; link; link = link->next)
        if (link->initialized && link->link_status) return link;
    return NULL;
}

/* ═══════ 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; }

/* Candidate TCP-пинг: по завершении (успех/ошибка/таймаут) зануляет свой слот в
 * mgr->cand_tcp_pings, чтобы conn_mgr_destroy не трогал уже освобождённый handle. */
static void cm_cand_tcp_ping_cb(int s, uint16_t r, void* arg, uint64_t n, const uint8_t* d, size_t l) {
    (void)s;(void)r;(void)n;(void)d;(void)l;
    struct stcp_client** slot = (struct stcp_client**)arg;
    if (slot) *slot = NULL;
}

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 ETCP_SOCKET* s = mgr->instance->etcp_sockets;
        while (s) { if (s->local_addr.ss_family == AF_INET6) break; s = s->next; }
        if (!s) break;
        if (cm_classify_v6_addr(a6->addr) == CM_V6_LL) sin6.sin6_scope_id = s->netif_index;
        struct sockaddr_storage sa; memcpy(&sa,&sin6,sizeof(sin6));
        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;
#ifdef UTUN_HAVE_STANDBY
    if (standby_get_sleep_tb() > 0) {
        mgr->bg_ping_wait = standby_wait(mgr, cm_bg_ping_timer_cb);
        return;
    }
    mgr->bg_ping_wait = NULL;
#endif
    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 = cm_active_direct_link(group,nq->node_id) != NULL;
            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) {
    return cm_active_direct_link(g,nid) != NULL;
}

/* Обновление кандидатов-посредников: удаляет протухших (>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();
#ifdef UTUN_HAVE_STANDBY
    if (standby_get_sleep_tb() > 0) {
        mgr->candidate_ping_wait = standby_wait(mgr, cm_candidate_ping_timer_cb);
        return;
    }
    mgr->candidate_ping_wait = NULL;
#endif
    /* Кандидаты-посредники выводятся из topo_node-списка группы: добавляем все
     * узлы с активным соединением (посредник должен быть напрямую связан с нами). */
    {
        struct ll_entry* e = mgr->group->nodes ? mgr->group->nodes->head : NULL;
        while (e) {
            struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e;
            if (nq->node_id != mgr->instance->node_id && cm_has_active_conn(mgr->group, nq->node_id)) {
                uint16_t rtt = topo_get_chain_rtt(nq);
                if (rtt == 0 || rtt == 0xFFFF) rtt = 1;
                conn_mgr_update_best_candidates(mgr, nq->node_id, rtt);
            }
            e = e->next;
        }
    }
    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_has_active_conn(mgr->group, nid)
            && 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; }
        struct ETCP_LINK* link = cm_active_direct_link(mgr->group,nid);
        if (link) {
            struct ETCP_CONN* dc = link->etcp;
            struct TOPO_NODE* ni = topo_node_registry_find(mgr->instance->topo_groups, nq->node_id);
            /* пир в SLEEP-фазе — прямое соединение живо, пинг будит его: пропускаем */
            if (ni && !dc->peer_sleep_phase) {
                DEBUG_TRACE(DEBUG_CATEGORY_BGP, "candidate probe: group=%016llx peer=%016llx link=%u transport=%s",
                            (unsigned long long)mgr->group->group_id, (unsigned long long)nid, link->local_link_id,
                            link->is_tcp ? "TCP" : "UDP");
                if (link->is_tcp) {
                    struct stcp_client** slot = NULL;
                    for (uint8_t i = 0; i < CONN_MGR_MAX_CANDIDATES; i++)
                        if (!mgr->cand_tcp_pings[i]) { slot = &mgr->cand_tcp_pings[i]; break; }
                    if (slot) etcp_send_tcp_ping(mgr->instance, ni->public_key, &link->remote_addr, CONN_PROBE_TIMEOUT_MS, cm_cand_tcp_ping_cb, slot, slot, link->reality_set ? &link->reality : NULL);
                } else
                    etcp_send_ping(mgr->instance, ni->public_key, &link->remote_addr, CONN_PROBE_TIMEOUT_MS, cm_bg_ping_noop_cb, NULL, NULL, 0);
            }
        } else route_connectivity_probe_node(mgr->instance, mgr->group, 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");
}