/** * @file topo_group.c * @brief BGP-like обмен топологией узлов (группы) */ #include #include #include #include "../lib/platform_compat.h" #include "../lib/debug_config.h" #include "../lib/mem.h" #include "../lib/memory_pool.h" #include "utun_instance.h" #include "etcp_api.h" #include "etcp.h" #include "etcp_connections.h" #include "etcp_debug.h" #include "config_parser.h" #include "topo_node.h" #include "topo_node_sqlite.h" #include "route_lib.h" #include "topo_group.h" #include "nat_detection.h" #include "route_connectivity.h" #include "control_server.h" #include "conn_mgr.h" #include "../transport_layer/node_conn_direct.h" #include "topo_group_connect.h" #include "topo_recovery.h" #include "broadcast.h" #include "../radio/radio.h" #include "../chat/chat_setting.h" #define TOPO_GROUP_UTUN 0x8000000000000000ULL // ============================================================================ // Вспомогательные функции // ============================================================================ struct TOPO_PEER_REQUEST { struct TOPO_GROUP_CONN_ITEM* peer; struct TOPO_PEER_REQUEST* next; }; static int topo_group_begin_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn, int retain); static int topo_group_peer_allowed(struct TOPO_GROUP* group, uint64_t node_id); static void topo_tx_clear(struct TOPO_GROUP_CONN_ITEM* peer); static int topo_tx_control(struct TOPO_GROUP_CONN_ITEM* peer, const void* data, size_t size); static int topo_tx_node(struct TOPO_GROUP_CONN_ITEM* peer, uint64_t node_id); static int topo_tx_table(struct TOPO_GROUP_CONN_ITEM* peer); static int topo_control_wait(struct ETCP_CONN* conn, const void* data, size_t size); static struct TOPO_GROUP_CONN_ITEM* topo_group_peer(const struct TOPO_GROUP* group, uint64_t peer_id) { for (struct ll_entry* e = group && group->senders_list ? group->senders_list->head : NULL; e; e = e->next) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data; if (item->node_id == peer_id) return item; } return NULL; } int topo_group_peer_ready(const struct TOPO_GROUP* group, uint64_t peer_id) { const struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, peer_id); return peer && peer->conn && peer->conn->links_up && !peer->conn->close_requested && peer->accepted && peer->table_received && peer->table_sent; } static void topo_group_peer_free(struct TOPO_GROUP_CONN_ITEM* peer) { topo_tx_clear(peer); while (peer->requests) { struct TOPO_PEER_REQUEST* request = peer->requests; peer->requests = request->next; request->peer = NULL; request->next = NULL; } if (peer->handle) node_conn_direct_close(peer->handle); topo_recovery_changed(peer->group); topo_group_connect_changed(peer->group); } static void topo_group_pending_remove(struct TOPO_GROUP_CONN_ITEM* peer) { struct ll_queue* queue = peer->group->senders_list; for (struct ll_entry* e = queue->head; e; e = e->next) { if ((void*)e->data != peer) continue; queue_remove_data(queue, e); topo_group_peer_free(peer); queue_entry_free(e); return; } } static void topo_group_peer_event(struct NODE_CONN_DIRECT* handle, enum ncd_event event, void* arg) { struct TOPO_GROUP_CONN_ITEM* peer = arg; if (event == NCD_EVENT_UP) { struct ETCP_CONN* conn = node_conn_direct_get_conn(handle); if (conn && topo_group_begin_conn(peer->group, conn, 0) == 0) return; DEBUG_WARN(DEBUG_CATEGORY_BGP, "group join failed peer=%016llx group=%016llx", (unsigned long long)peer->node_id, (unsigned long long)peer->group->group_id); } if (event == NCD_EVENT_TIMEOUT) DEBUG_WARN(DEBUG_CATEGORY_BGP, "group peer transport timeout peer=%016llx group=%016llx", (unsigned long long)peer->node_id, (unsigned long long)peer->group->group_id); if (peer->conn) topo_group_remove_conn(peer->group, peer->conn, TOPO_REMOVE_TRANSPORT_DOWN); else if ((event == NCD_EVENT_DOWN || event == NCD_EVENT_TIMEOUT) && peer->requests && node_conn_direct_get_conn(handle)) { peer->transport_down = 1; peer->retained = 0; topo_recovery_changed(peer->group); topo_group_connect_changed(peer->group); } else topo_group_pending_remove(peer); } int topo_group_peer_open(struct TOPO_GROUP* group, uint64_t node_id, struct TOPO_PEER_REQUEST** out) { if (out) *out = NULL; if (!group || !out || !node_id || !group->senders_list || group->stopping || !topo_group_peer_allowed(group, node_id)) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "cannot request group peer=%016llx: invalid arguments, stopped group or unknown member", (unsigned long long)node_id); return -1; } struct TOPO_PEER_REQUEST* request = u_calloc(1, sizeof(*request)); if (!request) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group request allocation failed"); return -1; } struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, node_id); if (!peer) { struct ll_entry* e = queue_entry_new(sizeof(*peer)); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group peer allocation failed"); u_free(request); return -1; } peer = (struct TOPO_GROUP_CONN_ITEM*)e->data; peer->group = group; peer->node_id = node_id; peer->progress = get_time_tb(); if (node_conn_direct_open(group->instance, node_id, topo_group_peer_event, peer, &peer->handle, NULL) == NCD_ERR) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "group NCD open failed peer=%016llx group=%016llx", (unsigned long long)node_id, (unsigned long long)group->group_id); queue_entry_free(e); u_free(request); return -1; } queue_data_put(group->senders_list, e); DEBUG_INFO(DEBUG_CATEGORY_BGP, "group peer CONNECTING peer=%016llx group=%016llx", (unsigned long long)node_id, (unsigned long long)group->group_id); } request->peer = peer; request->next = peer->requests; peer->requests = request; *out = request; return 0; } enum topo_peer_phase topo_group_peer_phase(const struct TOPO_PEER_REQUEST* request) { struct TOPO_GROUP_CONN_ITEM* peer = request ? request->peer : NULL; if (!peer || peer->tx_failed || peer->transport_down) return TOPO_PEER_FAILED; if (!peer->conn || !peer->conn->links_up) return TOPO_PEER_CONNECTING; return topo_group_peer_ready(peer->group, peer->node_id) ? TOPO_PEER_READY : TOPO_PEER_SYNCING; } struct ETCP_CONN* topo_group_peer_conn(const struct TOPO_PEER_REQUEST* request) { return request && request->peer ? node_conn_direct_get_conn(request->peer->handle) : NULL; } void topo_group_peer_leave(struct TOPO_GROUP* group, uint64_t node_id) { struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, node_id); if (!peer) return; if (peer->conn) topo_group_leave_peer(group, peer->handle); else topo_group_pending_remove(peer); } uint64_t topo_group_peer_progress(const struct TOPO_PEER_REQUEST* request) { return request && request->peer ? request->peer->progress : 0; } void topo_group_peer_close(struct TOPO_PEER_REQUEST* request) { if (!request) return; struct TOPO_GROUP_CONN_ITEM* peer = request->peer; if (peer) { struct TOPO_PEER_REQUEST** p = &peer->requests; while (*p && *p != request) p = &(*p)->next; if (*p) *p = request->next; if (!peer->requests && !peer->retained) { if (peer->conn) topo_group_leave_peer(peer->group, peer->handle); else topo_group_pending_remove(peer); } } u_free(request); } static void topo_group_peer_progressed(struct TOPO_GROUP* group, uint64_t node_id) { struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, node_id); if (peer) peer->progress = get_time_tb(); topo_recovery_changed(group); topo_group_connect_changed(group); } static void topo_group_log_exchange(struct TOPO_GROUP* group, struct TOPO_GROUP_CONN_ITEM* peer) { if (topo_group_peer_ready(group, peer->node_id)) { peer->retained = 1; topo_group_connect_on_up(group, peer->conn); } topo_group_peer_progressed(group, peer->node_id); DEBUG_INFO(DEBUG_CATEGORY_BGP, "group exchange: group=%016llx peer=%016llx exchange=%016llx sent=%u received=%u ready=%d", (unsigned long long)group->group_id, (unsigned long long)peer->conn->peer_node_id, (unsigned long long)peer->local_epoch, peer->table_sent, peer->table_received, topo_group_peer_ready(group, peer->conn->peer_node_id)); } static struct TOPOMSG_HEADER topo_header(struct TOPO_GROUP_CONN_ITEM* peer, uint8_t cmd) { return (struct TOPOMSG_HEADER){ .cmd = ETCP_ID_TOPO_ENTRY, .subcmd = cmd, .group_id = peer->group->group_id, .src_epoch = peer->local_epoch, .dst_epoch = peer->peer_epoch }; } static int topo_emit_control(struct ETCP_CONN* conn, const void* data, size_t size) { struct ll_entry* e = ll_alloc_lldgram(size); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group control allocation failed size=%zu", size); return -1; } memcpy(e->dgram, data, size); e->len = size; if (etcp_send(conn, e) == 0) return 0; DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group control send failed peer=%016llx cmd=%u", (unsigned long long)conn->peer_node_id, ((const uint8_t*)data)[1]); queue_dgram_free(e); queue_entry_free(e); return -1; } /* Сообщения сессии проходят одну FIFO. Stateless RESYNC/REJECT также ждут * пустой транспортной очереди, но не создают групповое участие. */ static int topo_send_control(struct ETCP_CONN* conn, const void* data, size_t size) { if (size >= sizeof(struct TOPOMSG_HEADER)) { const struct TOPOMSG_HEADER* h = data; struct TOPO_GROUP* group = topo_groups_find(conn->instance->topo_groups, h->group_id); struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (peer && peer->conn == conn && peer->local_epoch == h->src_epoch) return topo_tx_control(peer, data, size); } return topo_control_wait(conn, data, size); } static int topo_group_reset_exchange(struct TOPO_GROUP_CONN_ITEM* peer) { topo_tx_clear(peer); peer->tx_failed = 0; peer->transport_down = 0; peer->accepted = 0; peer->table_received = 0; peer->table_sent = 0; peer->peer_epoch = 0; uint64_t epoch; if (random_bytes((uint8_t*)&epoch, sizeof(epoch)) != 0 || !epoch) { peer->local_epoch = 0; DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot generate group epoch group=%016llx", (unsigned long long)peer->group->group_id); return -1; } peer->local_epoch = epoch; topo_group_peer_progressed(peer->group, peer->node_id); DEBUG_INFO(DEBUG_CATEGORY_BGP, "group session started: group=%016llx peer=%016llx local=%016llx", (unsigned long long)peer->group->group_id, (unsigned long long)peer->node_id, (unsigned long long)epoch); return 0; } static void topo_group_send_table_request(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (!peer || !peer->accepted) return; struct TOPOMSG_HEADER msg = topo_header(peer, TOPO_SUBCMD_REQUEST_TABLE); topo_send_control(conn, &msg, sizeof(msg)); } static void topo_group_send_join_group(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (!peer || peer->conn != conn || !peer->local_epoch) return; struct TOPOMSG_JOIN_GROUP msg = { .h = topo_header(peer, TOPO_SUBCMD_JOIN_GROUP), .group_type = group->group_type }; msg.h.dst_epoch = 0; topo_send_control(conn, &msg, sizeof(msg)); } static void topo_group_send_resync(struct ETCP_CONN* conn) { struct TOPOMSG_RESYNC msg = { .cmd = ETCP_ID_TOPO_ENTRY, .subcmd = TOPO_SUBCMD_RESYNC }; topo_send_control(conn, &msg, sizeof(msg)); } static int topo_group_send_table_complete(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (!peer || !peer->accepted) return -1; struct TOPOMSG_HEADER msg = topo_header(peer, TOPO_SUBCMD_TABLE_COMPLETE); return topo_emit_control(conn, &msg, sizeof(msg)); } static void topo_group_reject(struct ETCP_CONN* conn, const struct TOPOMSG_HEADER* h, enum topo_join_reject reason) { struct TOPOMSG_JOIN_GROUP msg = { .h = { .cmd = ETCP_ID_TOPO_ENTRY, .subcmd = TOPO_SUBCMD_JOIN_REJECT, .group_id = h->group_id, .dst_epoch = h->src_epoch }, .group_type = reason }; DEBUG_INFO(DEBUG_CATEGORY_BGP, "group JOIN rejected: group=%016llx peer=%016llx requested=%016llx reason=%u", (unsigned long long)h->group_id, (unsigned long long)conn->peer_node_id, (unsigned long long)h->src_epoch, reason); topo_send_control(conn, &msg, sizeof(msg)); } static int topo_group_add_to_senders(struct TOPO_GROUP* group, struct ETCP_CONN* conn); static bool topo_group_should_send_to(const struct TOPO_GROUP_NODE* nq, uint64_t target_id); static void topo_group_handle_request_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn); static void topo_group_handle_join_group(struct TOPO_GROUP* group, struct ETCP_CONN* conn, const struct TOPOMSG_JOIN_GROUP* msg); static void topo_group_handle_resync(struct UTUN_INSTANCE* instance, struct ETCP_CONN* conn); /* Краткий DEBUG-дамп принятого NODEINFO (id/ver/счётчики/hops). */ static void nodeinfo_dump_log(const uint8_t* data, size_t len) { if (!data || len < sizeof(struct TOPOMSG_NODEINFO_PKT)) return; struct TOPOMSG_NODEINFO_PKT* pkt = (struct TOPOMSG_NODEINFO_PKT*)data; struct TOPOMSG_NODE* ni = &pkt->node; DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "NODEINFO nid=%016llx ver=%u v4s=%u v4a=%u hops=%u", (unsigned long long)ni->node_id, ni->ver, ni->local_v4_sockets, ni->local_v4_addrs, ni->hop_count); } /* Форматирует NODEINFO в читаемую строку для лога (имя, подсети, адреса). */ static char* nodeinfo_format(const uint8_t* data, size_t len) { if (!data || len < sizeof(struct TOPOMSG_NODEINFO_PKT)) return NULL; struct TOPOMSG_NODEINFO_PKT* pkt = (struct TOPOMSG_NODEINFO_PKT*)data; struct TOPOMSG_NODE* ni = &pkt->node; uint64_t node_id = ni->node_id; const uint8_t* dyn = data + sizeof(struct TOPOMSG_NODEINFO_PKT); size_t off = 0; char name_buf[64] = {0}; size_t nl = ni->node_name_len; if (nl > 0 && nl < 63 && off + nl < len - sizeof(struct TOPOMSG_NODEINFO_PKT)) { memcpy(name_buf, dyn + off, nl); off += nl; } off += ni->local_v4_sockets * sizeof(struct TOPOMSG_SOCKMETA4); off += ni->local_v4_addrs * sizeof(struct TOPOMSG_ADDR4); off += ni->local_v6_sockets * sizeof(struct TOPOMSG_SOCKMETA6); off += ni->local_v6_addrs * sizeof(struct TOPOMSG_ADDR6); char subs_buf[512] = {0}; if (ni->local_v4_subnets > 0) { const struct TOPOMSG_SUBNET4* subs = (const struct TOPOMSG_SUBNET4*)(dyn + off); size_t sl = 0; for (uint8_t i = 0; i < ni->local_v4_subnets && sl < 400; i++) { uint32_t addr = (subs[i].addr[0]<<24)|(subs[i].addr[1]<<16)|(subs[i].addr[2]<<8)|subs[i].addr[3]; char tmp[64]; snprintf(tmp,sizeof(tmp),"%d.%d.%d.%d/%u ",(addr>>24)&255,(addr>>16)&255,(addr>>8)&255,addr&255,subs[i].prefix_length); strcat(subs_buf,tmp); sl += strlen(tmp); } } int need = snprintf(NULL,0,"NODEINFO nid=%016llx ver=%u name=\"%s\" v4subs=\"%s\" v4s=%u v4a=%u hop=%u", (unsigned long long)node_id,(unsigned)ni->ver,name_buf,subs_buf,(unsigned)ni->local_v4_sockets,(unsigned)ni->local_v4_addrs,(unsigned)ni->hop_count); char* buf = u_malloc(need+1); if (!buf) return NULL; snprintf(buf,need+1,"NODEINFO nid=%016llx ver=%u name=\"%s\" v4subs=\"%s\" v4s=%u v4a=%u hop=%u", (unsigned long long)node_id,(unsigned)ni->ver,name_buf,subs_buf,(unsigned)ni->local_v4_sockets,(unsigned)ni->local_v4_addrs,(unsigned)ni->hop_count); return buf; } /* Читаемое имя sub-команды BGP для логов. */ static const char* group_subcmd_name(uint8_t subcmd) { switch (subcmd) { case TOPO_SUBCMD_NODEINFO: return "NODEINFO"; case TOPO_SUBCMD_REQUEST_TABLE: return "REQ_TABLE"; case TOPO_SUBCMD_WITHDRAW: return "WITHDRAW"; case TOPO_SUBCMD_TABLE_COMPLETE: return "TABLE_COMPLETE"; case TOPO_SUBCMD_ERR_GROUP_MISMATCH: return "ERR_GROUP_MISMATCH"; case TOPO_SUBCMD_JOIN_GROUP: return "JOIN_GROUP"; case TOPO_SUBCMD_JOIN_ACCEPT: return "JOIN_ACCEPT"; case TOPO_SUBCMD_JOIN_REJECT: return "JOIN_REJECT"; case TOPO_SUBCMD_RESYNC: return "RESYNC"; case TOPO_SUBCMD_LEAVE_GROUP: return "LEAVE_GROUP"; default: return "?"; } } // ============================================================================ // Broadcast / Withdraw // ============================================================================ /* Рассылает WITHDRAW всем BGP-пирам (кроме exclude) об удалении узла. */ static void topo_group_broadcast_withdraw(struct TOPO_GROUP* group, uint64_t node_id, uint64_t wd_source, struct ETCP_CONN* exclude) { if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group is NULL"); return; } for (struct ll_entry* e = group->senders_list ? group->senders_list->head : NULL; e; e = e->next) { struct TOPO_GROUP_CONN_ITEM* peer = (struct TOPO_GROUP_CONN_ITEM*)e->data; if (!peer->conn || !peer->accepted || peer->conn == exclude) continue; struct TOPOMSG_WITHDRAW_PKT msg = { .h = topo_header(peer, TOPO_SUBCMD_WITHDRAW), .node_id = node_id, .wd_source = wd_source }; topo_send_control(peer->conn, &msg, sizeof(msg)); } } // ============================================================================ // Приём пакетов // ============================================================================ static int topo_configured(struct UTUN_INSTANCE* instance, uint64_t node_id) { for (struct CONFIG_CONN_HANDLE* owner = instance->config_conn_handles; owner; owner = owner->next) if (owner->node_id == node_id) return 1; return 0; } static int topo_group_has_sender(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { for (struct ll_entry* e = group->senders_list ? group->senders_list->head : NULL; e; e = e->next) if (((struct TOPO_GROUP_CONN_ITEM*)e->data)->conn == conn) return 1; return 0; } /* Транспорт не даёт членства: CHAT сверяет локальную таблицу участников. */ static int topo_group_peer_allowed(struct TOPO_GROUP* group, uint64_t node_id) { if (group->group_type != TOPO_GROUP_TYPE_CHAT) return 1; sqlite3* db = group->instance->topo_sqlite_db; return db && group->channel_id[0] && topo_node_sqlite_member_in_channel(db, group->channel_id, node_id); } /* Приёмник BGP-пакетов от ETCP: диспетчер по subcmd (NODEINFO/WITHDRAW/...). */ static void topo_group_receive_cbk(struct ETCP_CONN* from_conn, struct ll_entry* entry) { if (!entry) return; if (!from_conn || !entry->dgram || entry->len < 2) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "invalid BGP packet length=%u", entry->len); goto done; } struct UTUN_INSTANCE* instance = from_conn->instance; if (!instance || !instance->topo_groups) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid instance/topo_groups"); goto done; } const uint8_t* data = entry->dgram; uint8_t subcmd = data[1]; if (data[0] != ETCP_ID_TOPO_ENTRY) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "invalid BGP command=%u", data[0]); goto done; } if (subcmd == TOPO_SUBCMD_RESYNC && entry->len == sizeof(struct TOPOMSG_RESYNC)) { topo_group_handle_resync(instance, from_conn); goto done; } size_t size = sizeof(struct TOPOMSG_HEADER); if (subcmd == TOPO_SUBCMD_JOIN_GROUP || subcmd == TOPO_SUBCMD_JOIN_ACCEPT || subcmd == TOPO_SUBCMD_JOIN_REJECT) size = sizeof(struct TOPOMSG_JOIN_GROUP); else if (subcmd == TOPO_SUBCMD_NODEINFO) size = sizeof(struct TOPOMSG_NODEINFO_PKT); else if (subcmd == TOPO_SUBCMD_WITHDRAW) size = sizeof(struct TOPOMSG_WITHDRAW_PKT); else if (subcmd == TOPO_SUBCMD_ERR_GROUP_MISMATCH) size = sizeof(struct TOPOMSG_ERR_GROUP_MISMATCH); else if (subcmd != TOPO_SUBCMD_REQUEST_TABLE && subcmd != TOPO_SUBCMD_TABLE_COMPLETE && subcmd != TOPO_SUBCMD_LEAVE_GROUP) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "unknown BGP command=%u", subcmd); goto done; } if (entry->len < size || (subcmd != TOPO_SUBCMD_NODEINFO && entry->len != size)) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "invalid BGP length=%u cmd=%u expected=%zu", entry->len, subcmd, size); goto done; } const struct TOPOMSG_HEADER* h = (const struct TOPOMSG_HEADER*)data; struct TOPO_GROUP* group = topo_groups_find(instance->topo_groups, h->group_id); if (subcmd == TOPO_SUBCMD_JOIN_GROUP) { if (!h->src_epoch || h->dst_epoch) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "invalid JOIN epochs"); goto done; } if (!group || group->stopping) topo_group_reject(from_conn, h, TOPO_JOIN_NO_GROUP); else if (!topo_group_peer_allowed(group, from_conn->peer_node_id)) topo_group_reject(from_conn, h, TOPO_JOIN_NOT_MEMBER); else if (((const struct TOPOMSG_JOIN_GROUP*)data)->group_type != group->group_type) topo_group_reject(from_conn, h, TOPO_JOIN_WRONG_TYPE); else topo_group_handle_join_group(group, from_conn, (const struct TOPOMSG_JOIN_GROUP*)data); goto done; } struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, from_conn->peer_node_id); /* До получения JOIN ответа отменяем только ещё не согласованное предложение * с тем же src_epoch. У READY-сессии LEAVE всегда проверяет оба поколения. */ if (subcmd == TOPO_SUBCMD_LEAVE_GROUP && peer && peer->conn == from_conn && !peer->accepted && !h->dst_epoch && h->src_epoch && h->src_epoch == peer->peer_epoch) { topo_group_remove_conn(group, from_conn, TOPO_REMOVE_REMOTE_LEAVE); goto done; } if (!peer || peer->conn != from_conn || !h->dst_epoch || h->dst_epoch != peer->local_epoch) goto stale; if (subcmd == TOPO_SUBCMD_JOIN_REJECT) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "group JOIN refused: group=%016llx peer=%016llx local=%016llx reason=%u", (unsigned long long)h->group_id, (unsigned long long)peer->node_id, (unsigned long long)peer->local_epoch, ((const struct TOPOMSG_JOIN_GROUP*)data)->group_type); topo_group_remove_conn(group, from_conn, TOPO_REMOVE_REJECTED); goto done; } if (!h->src_epoch || (peer->peer_epoch && peer->peer_epoch != h->src_epoch)) goto stale; if (!topo_group_peer_allowed(group, from_conn->peer_node_id)) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "group peer no longer a member group=%016llx peer=%016llx", (unsigned long long)h->group_id, (unsigned long long)peer->node_id); goto done; } if (subcmd == TOPO_SUBCMD_JOIN_ACCEPT) { if (((const struct TOPOMSG_JOIN_GROUP*)data)->group_type != group->group_type) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "JOIN_ACCEPT type mismatch"); goto done; } if (!peer->accepted) { peer->peer_epoch = h->src_epoch; peer->accepted = 1; topo_group_peer_progressed(group, peer->node_id); DEBUG_INFO(DEBUG_CATEGORY_BGP, "group JOIN accepted: group=%016llx peer=%016llx local=%016llx remote=%016llx", (unsigned long long)h->group_id, (unsigned long long)peer->node_id, (unsigned long long)peer->local_epoch, (unsigned long long)peer->peer_epoch); topo_group_send_table_request(group, from_conn); } goto done; } if (subcmd == TOPO_SUBCMD_LEAVE_GROUP && peer->peer_epoch) { topo_group_remove_conn(group, from_conn, TOPO_REMOVE_REMOTE_LEAVE); goto done; } if (!peer->accepted || !peer->peer_epoch) goto stale; DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "BGP recv %s peer=%016llx group=%016llx local=%016llx remote=%016llx len=%u", group_subcmd_name(subcmd), (unsigned long long)peer->node_id, (unsigned long long)h->group_id, (unsigned long long)peer->local_epoch, (unsigned long long)peer->peer_epoch, entry->len); if (subcmd == TOPO_SUBCMD_NODEINFO) { nodeinfo_dump_log(data, entry->len); topo_group_process_nodeinfo(group, from_conn, data, entry->len); } else if (subcmd == TOPO_SUBCMD_WITHDRAW) topo_group_process_withdraw(group, from_conn, data, entry->len); else if (subcmd == TOPO_SUBCMD_REQUEST_TABLE) topo_group_handle_request_table(group, from_conn); else if (subcmd == TOPO_SUBCMD_TABLE_COMPLETE && !peer->table_received) { peer->table_received = 1; topo_group_log_exchange(group, peer); } else if (subcmd == TOPO_SUBCMD_ERR_GROUP_MISMATCH) { const struct TOPOMSG_ERR_GROUP_MISMATCH* err = (const struct TOPOMSG_ERR_GROUP_MISMATCH*)data; DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group mismatch peer=%016llx expected=%u received=%u", (unsigned long long)peer->node_id, err->expected_type, err->received_flags); } goto done; stale: DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "ignore unmatched group message cmd=%u group=%016llx peer=%016llx src=%016llx dst=%016llx", subcmd, (unsigned long long)h->group_id, (unsigned long long)from_conn->peer_node_id, (unsigned long long)h->src_epoch, (unsigned long long)h->dst_epoch); done: queue_dgram_free(entry); queue_entry_free(entry); } // ============================================================================ // Init / Destroy / New / Remove conn // ============================================================================ /* Callback статуса ETCP-соединения: UP — добавить пира в группы; DOWN — удалить. */ static void topo_group_conn_status(struct ETCP_CONN* conn, int status, void* arg) { struct TOPO_GROUPS* groups = (struct TOPO_GROUPS*)arg; if (!conn || !groups) return; if (status == ETCP_CONN_STATUS_REINIT) { // Сброс normalizer мог уничтожить TABLE_REQUEST/NODEINFO. Повторяем обмен, сохраняя членство и пути. for (struct ll_entry* e = groups->group_list->head; e; e = e->next) { struct TOPO_GROUP* g = (struct TOPO_GROUP*)e; for (struct ll_entry* se = g->senders_list->head; se; se = se->next) { struct TOPO_GROUP_CONN_ITEM* peer = (struct TOPO_GROUP_CONN_ITEM*)se->data; if (peer->conn != conn) continue; if (topo_group_reset_exchange(peer) < 0) continue; DEBUG_INFO(DEBUG_CATEGORY_BGP,"BGP session resync: peer=%016llx group=%016llx", (unsigned long long)conn->peer_node_id,(unsigned long long)g->group_id); topo_group_send_join_group(g, conn); break; } } return; } if (status == ETCP_CONN_STATUS_UP && conn->peer_node_id) { /* UTUN присоединяет пир через UP принадлежащего ему NCD handle. * Сам по себе транспортный UP не означает участия в UTUN-группе. */ if (!topo_configured(conn->instance, conn->peer_node_id)) topo_group_send_resync(conn); /* CHAT: только члены каналов (peers_* в БД). */ sqlite3* db = conn->instance ? conn->instance->topo_sqlite_db : NULL; uint64_t* chs = NULL; int chn = 0; if (db && topo_node_sqlite_get_member_channels(db, conn->peer_node_id, &chs, &chn) == 0 && chn > 0) { for (int i = 0; i < chn; i++) { struct TOPO_GROUP* cg = topo_groups_find(groups, chs[i]); if (cg && cg->group_type == TOPO_GROUP_TYPE_CHAT) topo_group_begin_conn(cg, conn, 0); } } u_free(chs); } else if (status == ETCP_CONN_STATUS_DOWN || status == ETCP_CONN_STATUS_DELETE) { struct ll_entry* fe = groups->group_list->head; while (fe) { struct TOPO_GROUP* g = (struct TOPO_GROUP*)fe; fe = fe->next; topo_group_remove_conn(g, conn, TOPO_REMOVE_TRANSPORT_DOWN); } } } /* Создаёт группу: очереди nodes/senders, local_node, conn_mgr, broadcast. */ static struct TOPO_GROUP* topo_group_create(struct UTUN_INSTANCE* instance, uint64_t group_id, uint8_t group_type) { if (!instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "instance is NULL"); return NULL; } DEBUG_INFO(DEBUG_CATEGORY_BGP, "group_id=%016llx node_id=%016llx", (unsigned long long)group_id, (unsigned long long)instance->node_id); struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_GROUP)); if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "alloc failed"); return NULL; } struct TOPO_GROUP* group = (struct TOPO_GROUP*)qe; memset((uint8_t*)group + sizeof(struct ll_entry), 0, sizeof(*group) - sizeof(struct ll_entry)); group->group_id = group_id; group->group_type = group_type; group->instance = instance; sc_derive_ed25519_pubkey(instance->my_keys.private_key, group->ed25519_public_key); DEBUG_INFO(DEBUG_CATEGORY_BGP, "Ed25519 pubkey derived from X25519 private key"); group->nodes = queue_new(instance->ua, BGP_NODES_HASH_SIZE, offsetof(struct TOPO_GROUP_NODE, node_id) - sizeof(struct ll_entry), 8, "group_nodes"); if (!group->nodes) { queue_entry_free(qe); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "nodes queue creation failed"); return NULL; } group->senders_list = queue_new(instance->ua, 0, 0, 0, "group_senders"); if (!group->senders_list) { queue_free(group->nodes); queue_entry_free(qe); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "senders_list queue creation failed"); return NULL; } group->local_node = NULL; topo_group_update_my_nodeinfo(instance, group); group->conn_mgr = conn_mgr_init(group); broadcast_init(group); radio_init(group); DEBUG_INFO(DEBUG_CATEGORY_BGP, "TOPO group created: group_id=%016llx", (unsigned long long)group_id); return group; } /* Полностью разрушает группу: recovery, connect, broadcast, conn_mgr, узлы. */ static void topo_group_destroy(struct TOPO_GROUP* group) { if (!group) return; group->stopping = 1; DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 0 enter grp=%p id=0x%016llx rec_list=%p connect=%p", group, (unsigned long long)group->group_id, group->recovery, group->connect); topo_recovery_cancel_all(group); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 1 recovery_cancel done grp=%p", group); topo_group_connect_destroy(group); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 2 connect_destroy done"); broadcast_destroy(group); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 2b broadcast_destroy done"); radio_destroy(group); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 2c radio_destroy done"); struct ll_entry* e; while ((e = queue_data_get(group->senders_list)) != NULL) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data; DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "group destroy: release peer=%016llx group=%016llx", (unsigned long long)node_conn_direct_node_id(item->handle), (unsigned long long)group->group_id); topo_group_peer_free(item); queue_entry_free(e); } queue_free(group->senders_list); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 3 senders_list done"); if (group->conn_mgr) { conn_mgr_destroy(group->conn_mgr); group->conn_mgr = NULL; } DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 4 conn_mgr done"); if (group->nodes) { int nc = 0; struct ll_entry* e = group->nodes->head; while (e) { struct ll_entry* next = e->next; topo_nodeq_remove_node(group, (struct TOPO_GROUP_NODE*)e); e = next; nc++; } queue_free(group->nodes); if (nc) DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 4b nodes cleaned (%d)", nc); } if (group->local_node) { if (group->local_node->paths) { queue_free(group->local_node->paths); group->local_node->paths = NULL; } topo_nodeq_free_group_fields(group->instance->topo_groups, group->local_node); u_free(group->local_node); } /* освободить подписчиков BGP-событий (member_sync и др.) */ while (group->node_cbks) { struct topo_node_cbk_entry* c = group->node_cbks; group->node_cbks = c->next; u_free(c); } DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 5 done"); } /* Удаляет группу из контейнера (кроме UTUN-группы по умолчанию). */ void topo_groups_remove_group(struct TOPO_GROUPS* g, uint64_t group_id) { if (!g || !g->group_list || group_id == TOPO_GROUP_UTUN) return; struct TOPO_GROUP* group = topo_groups_find(g, group_id); if (!group) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "remove_group: group %016llx not found", (unsigned long long)group_id); return; } DEBUG_INFO(DEBUG_CATEGORY_BGP, "remove_group: id=%016llx type=%d ch=%s", (unsigned long long)group_id, group->group_type, group->channel_id); queue_remove_data(g->group_list, &group->ll); topo_group_destroy(group); /* node_cbks освобождаются внутри */ queue_entry_free(&group->ll); } /* Инициализация контейнера групп: реестр, пулы, SQLite, default-группа, подписки. */ struct TOPO_GROUPS* topo_groups_init(struct UTUN_INSTANCE* instance) { if (!instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "instance is NULL"); return NULL; } DEBUG_INFO(DEBUG_CATEGORY_BGP, "node_id=%016llx", (unsigned long long)instance->node_id); struct TOPO_GROUPS* g = u_calloc(1, sizeof(struct TOPO_GROUPS)); if (!g) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "alloc failed"); return NULL; } g->instance = instance; g->group_list = queue_new(instance->ua, 16, 0, 8, "topo_groups"); if (!g->group_list) { u_free(g); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group_list queue creation failed"); return NULL; } g->node_registry = queue_new(instance->ua, TOPO_NODE_REGISTRY_HASH_SIZE, 0, 8, "node_registry"); if (!g->node_registry) { queue_free(g->group_list); u_free(g); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "node_registry queue creation failed"); return NULL; } g->v4_sock_meta_pool = memory_pool_init(sizeof(struct TOPO_SOCKMETA4), "to_v4sock"); g->v4_addr_pool = memory_pool_init(sizeof(struct TOPO_ADDR4), "to_v4addr"); g->v6_sock_meta_pool = memory_pool_init(sizeof(struct TOPO_SOCKMETA6), "to_v6sock"); g->v6_addr_pool = memory_pool_init(sizeof(struct TOPO_ADDR6), "to_v6addr"); g->v4_subnet_pool = memory_pool_init(sizeof(struct TOPO_SUBNET4), "to_v4sub"); g->v6_subnet_pool = memory_pool_init(sizeof(struct TOPO_SUBNET6), "to_v6sub"); if (instance->config && instance->config->global.db_path[0]) { char db_file[512]; snprintf(db_file, sizeof(db_file), "%s/chats.db", instance->config->global.db_path); int rc = sqlite3_open_v2(db_file, &instance->topo_sqlite_db, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX, NULL); if (rc == SQLITE_OK && instance->topo_sqlite_db) { sqlite3_exec(instance->topo_sqlite_db, "PRAGMA journal_mode=WAL", NULL, NULL, NULL); sqlite3_exec(instance->topo_sqlite_db, "PRAGMA foreign_keys=ON", NULL, NULL, NULL); sqlite3_exec(instance->topo_sqlite_db, "PRAGMA synchronous=NORMAL", NULL, NULL, NULL); sqlite3_exec(instance->topo_sqlite_db, "PRAGMA wal_autocheckpoint=10000", NULL, NULL, NULL); sqlite3_exec(instance->topo_sqlite_db, "PRAGMA cache_size=-32768", NULL, NULL, NULL); sqlite3_exec(instance->topo_sqlite_db, "PRAGMA mmap_size=134217728", NULL, NULL, NULL); sqlite3_exec(instance->topo_sqlite_db, "PRAGMA temp_store=MEMORY", NULL, NULL, NULL); topo_node_sqlite_init(instance->topo_sqlite_db); DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "SQLite opened: %s rc=%d db=%p", db_file, rc, (void*)instance->topo_sqlite_db); } else { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "sqlite3_open(%s) failed: %s", db_file, instance->topo_sqlite_db ? sqlite3_errmsg(instance->topo_sqlite_db) : "null db"); if (instance->topo_sqlite_db) { sqlite3_close(instance->topo_sqlite_db); instance->topo_sqlite_db = NULL; } } } instance->topo_groups = g; struct TOPO_GROUP* default_group = topo_group_create(instance, TOPO_GROUP_UTUN, TOPO_GROUP_TYPE_UTUN); if (!default_group) { memory_pool_destroy(g->v4_sock_meta_pool); memory_pool_destroy(g->v4_addr_pool); memory_pool_destroy(g->v6_sock_meta_pool); memory_pool_destroy(g->v6_addr_pool); memory_pool_destroy(g->v4_subnet_pool); memory_pool_destroy(g->v6_subnet_pool); queue_free(g->node_registry); queue_free(g->group_list); u_free(g); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "failed to create default group"); return NULL; } queue_data_put_with_index(g->group_list, &default_group->ll); instance->conn_mgr = default_group->conn_mgr; /* recv handler shortcut */ etcp_bind(instance, ETCP_ID_TOPO_ENTRY, topo_group_receive_cbk); etcp_add_conn_status_cbk(instance, topo_group_conn_status, g); etcp_add_socket_cbk(instance, topo_node_on_socket_changed, NULL, ETCP_SOCKET_EVENT_ADDR_CHANGED | ETCP_SOCKET_EVENT_STATUS_CHANGED); utun_add_activity_cbk(instance, topo_group_on_activity_change, NULL); DEBUG_INFO(DEBUG_CATEGORY_BGP, "TOPO groups initialized (with default group)"); return g; } /* Callback смены active/standby: обновляет свой nodeinfo и рассылает его пирам. */ void topo_group_on_activity_change(struct UTUN_INSTANCE* instance, int active, void* arg) { (void)active; (void)arg; if (!instance || !instance->topo_groups || !instance->topo_groups->group_list) return; struct ll_entry* ge = instance->topo_groups->group_list->head; while (ge) { struct TOPO_GROUP* g = (struct TOPO_GROUP*)ge; topo_group_update_my_nodeinfo(instance, g); if (g->senders_list) { struct ll_entry* se = g->senders_list->head; while (se) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data; if (item && item->conn) topo_group_send_nodeinfo(g, g->local_node, item->conn); se = se->next; } } ge = ge->next; } } /* Полный teardown контейнера: отписки, уничтожение групп, реестра, пулов. */ void topo_groups_destroy(struct UTUN_INSTANCE* instance) { if (!instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "instance is NULL"); return; } if (!instance->topo_groups) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "topo_groups is NULL"); return; } DEBUG_INFO(DEBUG_CATEGORY_BGP, "node_id=%016llx tgs=%p", (unsigned long long)instance->node_id, instance->topo_groups); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 0 enter"); etcp_unbind(instance, ETCP_ID_TOPO_ENTRY); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 1 etcp_unbind done"); etcp_remove_conn_status_cbk(instance, topo_group_conn_status, instance->topo_groups); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 2 conn_cbk_remove done"); etcp_remove_socket_cbk(instance, topo_node_on_socket_changed, NULL); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 3 socket_cbk_remove done"); utun_remove_activity_cbk(instance, topo_group_on_activity_change, NULL); route_connectivity_cancel_all(instance); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 4 connectivity_cancel done"); struct TOPO_GROUPS* g = instance->topo_groups; struct ll_entry* ge; int gcount = 0; while ((ge = queue_data_get(g->group_list)) != NULL) { struct TOPO_GROUP* grp = (struct TOPO_GROUP*)ge; topo_group_destroy(grp); queue_entry_free(ge); gcount++; } queue_free(g->group_list); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 5 groups destroyed (%d)", gcount); if (g->node_registry) { int ncount = 0; struct ll_entry* re; while ((re = queue_data_get(g->node_registry)) != NULL) { struct TOPO_NODE* node; memcpy(&node, re->data + 8, sizeof(node)); topo_node_destroy(g, node); queue_entry_free(re); ncount++; } queue_free(g->node_registry); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 6 node_registry done (%d)", ncount); } else { DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 6 node_registry skip (NULL)"); } memory_pool_destroy(g->v4_sock_meta_pool); memory_pool_destroy(g->v4_addr_pool); memory_pool_destroy(g->v6_sock_meta_pool); memory_pool_destroy(g->v6_addr_pool); memory_pool_destroy(g->v4_subnet_pool); memory_pool_destroy(g->v6_subnet_pool); DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 7 pools destroyed"); /* Shared SQLite DB закрывается в utun_instance_destroy() в самом конце — * chat/media/db_sync ещё используют его во время teardown. */ DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 8 sqlite deferred (закрывается в utun_instance_destroy)"); u_free(g); instance->topo_groups = NULL; instance->conn_mgr = NULL; DEBUG_INFO(DEBUG_CATEGORY_SYS, "[TOPO_DESTROY] 9 done"); } /* Возвращает UTUN-группу по умолчанию (group_id=TOPO_GROUP_UTUN). */ struct TOPO_GROUP* topo_groups_get_default(struct TOPO_GROUPS* g) { return topo_groups_find(g, TOPO_GROUP_UTUN); } /* Ищет группу по group_id (NULL, если нет). */ struct TOPO_GROUP* topo_groups_find(struct TOPO_GROUPS* g, uint64_t group_id) { if (!g || !g->group_list) return NULL; struct ll_entry* e = queue_find_data_by_index(g->group_list, &group_id); return e ? (struct TOPO_GROUP*)e : NULL; } /* Создаёт и регистрирует новую группу; добавляет существующие conn, для CHAT — автоподключение. */ struct TOPO_GROUP* topo_groups_create_group(struct TOPO_GROUPS* g, uint64_t group_id, uint8_t group_type, const char* channel_id) { if (!g || !g->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid args"); return NULL; } if (topo_groups_find(g, group_id)) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group_id=%016llx already exists", (unsigned long long)group_id); return NULL; } struct TOPO_GROUP* group = topo_group_create(g->instance, group_id, group_type); if (!group) return NULL; if (channel_id) { strncpy(group->channel_id, channel_id, sizeof(group->channel_id) - 1); group->channel_id[sizeof(group->channel_id) - 1] = '\0'; } queue_data_put_with_index(g->group_list, &group->ll); struct ll_queue* connections = g->instance->connections; if (connections) { sqlite3* db = g->instance->topo_sqlite_db; for (uint32_t slot = 0; slot < connections->hash_size; slot++) { struct ll_entry* entry = connections->hash_table[slot]; while (entry) { struct conn_queue_entry* cqe = (struct conn_queue_entry*)entry->data; if (cqe->conn && cqe->peer_node_id != 0) { if (group_type == TOPO_GROUP_TYPE_CHAT && db) { /* CHAT: добавляем conn только если пир — член именно этого канала */ uint64_t* chs = NULL; int chn = 0; if (topo_node_sqlite_get_member_channels(db, cqe->peer_node_id, &chs, &chn) == 0) { for (int i = 0; i < chn; i++) { if (chs[i] == group_id) { topo_group_new_conn(group, cqe->conn); break; } } } u_free(chs); } else if (group_type == TOPO_GROUP_TYPE_UTUN && cqe->conn->links_up) { if (topo_configured(g->instance, cqe->peer_node_id)) topo_group_new_conn(group, cqe->conn); } } entry = entry->hash_next; } } } DEBUG_INFO(DEBUG_CATEGORY_BGP, "Group created: group_id=%016llx type=%u ch_id=%s", (unsigned long long)group_id, group_type, group->channel_id); if (group_type == TOPO_GROUP_TYPE_CHAT && chat_setting_get_int(group->instance, "group_autoconnect", 1)) topo_group_connect_init(group); return group; } /* Разрешить/запретить NAT-проверку локальных подсетей (делегат в nat_detection). */ void topo_group_set_nat_check_local(struct TOPO_GROUP* group, int allow) { if (!group || !group->instance || !group->instance->nat_det) return; nat_detection_set_allow_local(group->instance->nat_det, allow); } /* ── Рация (PTT walkie-talkie, per-CHAT-группа) ── */ void topo_group_set_radio(struct TOPO_GROUP* group, int on) { if (!group || !group->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "set_radio: invalid args"); return; } if (group->group_type != TOPO_GROUP_TYPE_CHAT) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "set_radio: only CHAT groups support radio grp=%016llx type=%u", (unsigned long long)group->group_id, group->group_type); return; } uint8_t v = on ? 1 : 0; if (group->radio_active == v) { DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "set_radio: unchanged (%u) grp=%016llx ch=%s", v, (unsigned long long)group->group_id, group->channel_id); return; } group->radio_active = v; DEBUG_INFO(DEBUG_CATEGORY_BGP, "set_radio: %s grp=%016llx ch=%s", v ? "ON" : "OFF", (unsigned long long)group->group_id, group->channel_id); topo_group_update_my_nodeinfo(group->instance, group); if (group->senders_list) { struct ll_entry* se = group->senders_list->head; while (se) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data; if (item && item->conn && group->local_node) topo_group_send_nodeinfo(group, group->local_node, item->conn); se = se->next; } } } int topo_group_radio_subscribers(struct TOPO_GROUP* group) { if (!group || !group->nodes) return 0; int n = 0; struct ll_entry* e = group->nodes->head; while (e) { struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e; if (nq->radio && nq->node_id != group->instance->node_id) n++; e = e->next; } return n; } /* Устанавливает колбэк сохранения узла в БД (используется chatgui/member_sync). */ void topo_groups_set_node_updated_cb(struct TOPO_GROUPS* groups, topo_node_updated_fn fn) { if (groups) groups->node_updated_cb = fn; } /* Добавляет пира в группу при UP: дедуп, добавление в senders, запрос таблицы. */ static int topo_group_begin_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn, int retain) { if (!group || !conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid args"); return -1; } if (group->stopping) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "join on stopped group"); return -1; } if (!conn->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "conn->instance is NULL"); return -1; } if (!conn->instance->rt) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "conn->instance->rt is NULL"); return -1; } if (!topo_group_peer_allowed(group, conn->peer_node_id)) { DEBUG_INFO(DEBUG_CATEGORY_BGP, "group connect deferred: member unknown peer=%016llx group=%016llx", (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id); return -1; } struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (peer && retain) peer->retained = 1; /* дедуп: тот же conn стреляет ETCP_CONN_STATUS_UP дважды (UDP-линк, затем TCP-линк). * Если conn уже в senders_list — повторно не обрабатываем, иначе active_conn_count * задваивается и переподключение никогда не стартует. */ { struct ll_entry* se = group->senders_list ? group->senders_list->head : NULL; while (se) { if (((struct TOPO_GROUP_CONN_ITEM*)se->data)->conn == conn) { DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_new_conn: conn already in senders_list, skip double UP (%s)", conn->log_name); return 0; } se = se->next; } } struct TOPO_GROUP_NODE* peer_nq = topo_node_find_by_id(group, conn->peer_node_id); if (peer_nq) { peer_nq->connectivity.last_ping_time = get_time_tb(); peer_nq->conn_presence |= NCONN_DIRECT; peer_nq->conn_up |= NCONN_DIRECT; if (conn->instance->topo_sqlite_db) topo_node_sqlite_nodeinfo_updated(conn->instance->topo_sqlite_db, conn->peer_node_id); } DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_new_conn: ADD conn=%p peer=%016llx state=%d grp=%016llx ch=%s", (void*)conn, (unsigned long long)conn->peer_node_id, conn->state, (unsigned long long)group->group_id, group->channel_id); if (topo_group_add_to_senders(group, conn) != 0) return -1; topo_group_peer_progressed(group, conn->peer_node_id); DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "topo_group_new_conn: peer=%016llx group=%016llx type=%d ch=%s", (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id, group->group_type, group->channel_id); peer = topo_group_peer(group, conn->peer_node_id); if (topo_group_reset_exchange(peer) < 0) return -1; topo_group_send_join_group(group, conn); topo_group_connect_on_up(group, conn); return 0; } int topo_group_new_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { return topo_group_begin_conn(group, conn, 1); } /* Обрабатывает DOWN пира: чистит пути, каскадно удаляет узлы, шлёт withdraw. */ void topo_group_remove_conn(struct TOPO_GROUP* group, struct ETCP_CONN* conn, enum topo_group_remove_reason reason) { if (!group || !conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid args"); return; } if (!conn->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "invalid instance"); return; } struct TOPO_GROUP_CONN_ITEM* pending = topo_group_peer(group, conn->peer_node_id); if (pending && !pending->conn && node_conn_direct_get_conn(pending->handle) == conn) { DEBUG_INFO(DEBUG_CATEGORY_BGP, "group pending peer removed: peer=%016llx group=%016llx reason=%d", (unsigned long long)pending->node_id, (unsigned long long)group->group_id, reason); if (reason == TOPO_REMOVE_TRANSPORT_DOWN && pending->requests) { pending->transport_down = 1; pending->retained = 0; topo_recovery_changed(group); topo_group_connect_changed(group); } else topo_group_pending_remove(pending); return; } bool found_in_list = false; struct ll_entry* e = group->senders_list->head; while (e) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data; if (item->conn == conn) { found_in_list = true; break; } e = e->next; } if (!found_in_list) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "topo_group_remove_conn: conn=%p peer=%016llx state=%d NOT in senders_list (already removed?), skip grp=%016llx ch=%s", (void*)conn, (unsigned long long)conn->peer_node_id, conn->state, (unsigned long long)group->group_id, group->channel_id); return; } DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_remove_conn: conn=%p peer=%016llx state=%d name='%s' grp=%016llx ch=%s", (void*)conn, (unsigned long long)conn->peer_node_id, conn->state, conn->log_name, (unsigned long long)group->group_id, group->channel_id); struct ROUTE_TABLE* rt = conn->instance->rt; int nodes_removed = 0; int cascaded = 0; struct ll_entry* node_entry = group->nodes ? group->nodes->head : NULL; while (node_entry) { struct ll_entry* next = node_entry->next; struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)node_entry; { uint64_t next_hop = nq->node_id; uint16_t lost_rtt = UINT16_MAX; for (struct ll_entry* pe = nq->paths ? nq->paths->head : NULL; pe; pe = pe->next) { struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)pe; if (path->conn != conn) continue; /* links_up уже сброшен; сохраняем последний RTT до удаления пути. */ if (conn->rtt_last) { uint32_t total = (uint32_t)conn->rtt_last + path->cumulative_rtt; lost_rtt = total > UINT16_MAX ? UINT16_MAX : (uint16_t)total; } if (path->hop_count >= 2) { uint64_t* hop = (uint64_t*)((uint8_t*)path + sizeof(*path)); if (hop[path->hop_count - 1] == conn->peer_node_id) next_hop = hop[path->hop_count - 2]; } break; } if (topo_group_remove_path(nq, conn) == 1) { uint64_t key = nq->node_id; if (reason == TOPO_REMOVE_TRANSPORT_DOWN && key != conn->peer_node_id) { topo_recovery_add_node(group, key, next_hop, lost_rtt); cascaded++; } nq->conn_presence = 0; nq->conn_up = 0; topo_fire_nodeinfo_cbk(conn->instance, group, nq); if (group->group_type != TOPO_GROUP_TYPE_CHAT && rt) route_delete(rt, nq); if (conn->instance && conn->instance->control_srv) control_server_notify_node_removed(conn->instance->control_srv, key); topo_nodeq_remove_node(group, nq); nodes_removed++; DEBUG_INFO(DEBUG_CATEGORY_BGP, "node removed from group: node=%016llx grp=%016llx type=%u ch=%s (link_down)", (unsigned long long)key, (unsigned long long)group->group_id, group->group_type, group->channel_id); // REINIT временно убирает путь, но end-to-end данные роутера остаются неподтверждёнными. if (!conn->reinit_pending) etcp_router_conn_close_all_for_node(group->instance, group->group_id, key); uint64_t wd_src = (key == conn->peer_node_id) ? conn->instance->node_id : conn->peer_node_id; topo_group_broadcast_withdraw(group, key, wd_src, NULL); DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "Removed node %016llx after link down", (unsigned long long)key); { struct topo_node_cbk_entry* c = group->node_cbks; while (c) { c->fn(group, key, TOPO_NODE_EVENT_REMOVE, c->arg); c = c->next; } } } else { nq->conn_up &= ~NCONN_DIRECT; { int has_direct = 0; struct ll_entry* pe2 = nq->paths ? nq->paths->head : NULL; while (pe2) { struct TOPO_NODEPATH* p = (struct TOPO_NODEPATH*)pe2; if (p->conn->peer_node_id == nq->node_id) { has_direct = 1; break; } pe2 = pe2->next; } if (!has_direct) nq->conn_presence &= ~NCONN_DIRECT; } topo_fire_nodeinfo_cbk(conn->instance, group, nq); } } node_entry = next; } /* освобождаем владение BGP: сначала убираем item из senders_list (чтобы ре-ентрантный remove_conn от conn DOWN не нашёл его), затем закрываем handle. */ e = group->senders_list->head; while (e) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data; if (item->conn == conn) { if (reason == TOPO_REMOVE_TRANSPORT_DOWN && item->requests && node_conn_direct_get_conn(item->handle) == conn) { topo_tx_clear(item); item->conn = NULL; item->transport_down = 1; item->retained = 0; item->accepted = 0; item->table_sent = 0; item->table_received = 0; topo_recovery_changed(group); topo_group_connect_changed(group); } else { queue_remove_data(group->senders_list, e); topo_group_peer_free(item); queue_entry_free(e); } DEBUG_INFO(DEBUG_CATEGORY_BGP, "topo_group_remove_conn: REMOVED conn=%p peer=%016llx from senders_list grp=%016llx ch=%s", (void*)conn, (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id, group->channel_id); break; } e = e->next; } if (cascaded > 0) topo_recovery_start(group); topo_group_connect_on_down(group, conn); DEBUG_INFO(DEBUG_CATEGORY_BGP, "BGP peer removed: %s nodes=%d grp=%016llx reason=%d recovery=%d", conn->log_name, nodes_removed, (unsigned long long)group->group_id, reason, cascaded); } // ============================================================================ // NODEINFO BASED IMPLEMENTATION // ============================================================================ /* Ищет лучшее ETCP-соединение до узла (мин. hop_count среди живых путей). */ struct ETCP_CONN* topo_group_find_conn_for_node(struct TOPO_GROUP* group, uint64_t node_id) { if (!group) return NULL; struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(group, node_id); if (!nq || !nq->paths || !nq->paths->head) return NULL; struct ll_entry* e = nq->paths->head; struct TOPO_NODEPATH* best_live = NULL; uint8_t best_live_hops = 255; struct TOPO_NODEPATH* best_any = NULL; uint8_t best_any_hops = 255; while (e) { struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)e; if (path->conn) { if (path->conn->links_up && path->hop_count < best_live_hops) { best_live = path; best_live_hops = path->hop_count; } if (path->hop_count < best_any_hops) { best_any = path; best_any_hops = path->hop_count; } } e = e->next; } if (!best_live && best_any) DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "find_conn_for_node %016llx: no live paths, fallback to best (hops=%d links_up=%d)", (unsigned long long)node_id, best_any_hops, best_any->conn->links_up); return best_live ? best_live->conn : (best_any ? best_any->conn : NULL); } /* Добавляет путь (conn + hop_list + rtt) в paths узла. */ int topo_group_add_path(struct TOPO_GROUP_NODE* nq, struct ETCP_CONN* conn, uint64_t* hop_list, uint8_t hop_count, uint16_t cumulative_rtt) { if (!nq || !conn || hop_count > MAX_HOPS || !hop_list) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "add_path: invalid args"); return -1; } if (!nq->paths) { nq->paths = queue_new(conn->instance->ua, 0, 0, 0, "node_paths"); if (!nq->paths) return -1; } size_t path_size = sizeof(struct TOPO_NODEPATH) - sizeof(struct ll_entry) + hop_count * 8; struct ll_entry* pe = queue_entry_new(path_size); if (!pe) return -1; struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)pe; path->conn = conn; path->hop_count = hop_count; path->cumulative_rtt = cumulative_rtt; memcpy((uint64_t*)((uint8_t*)path + sizeof(struct TOPO_NODEPATH)), hop_list, hop_count * 8); queue_data_put(nq->paths, pe); return 0; } /* Удаляет пути, содержащие wd_source в hop_list; 1 — если путей не осталось. */ static int topo_group_remove_path_by_hop(struct TOPO_GROUP_NODE* nq, uint64_t wd_source) { if (!nq || !nq->paths) return 0; int removed = 0; struct ll_entry* e = nq->paths->head; while (e) { struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)e; uint64_t* hop = (uint64_t*)((uint8_t*)path + sizeof(struct TOPO_NODEPATH)); bool has_wd = false; for (uint8_t i = 0; i < path->hop_count; i++) if (hop[i] == wd_source) { has_wd = true; break; } if (has_wd) { struct ll_entry* next = e->next; queue_remove_data(nq->paths, e); queue_entry_free(e); removed++; e = next; continue; } e = e->next; } if (removed > 0 && nq->paths && queue_entry_count(nq->paths) == 0) { queue_free(nq->paths); nq->paths = NULL; return 1; } return removed; } /* Удаляет путь через конкретный conn; 1 — если путей не осталось. */ int topo_group_remove_path(struct TOPO_GROUP_NODE* nq, struct ETCP_CONN* conn) { if (!nq || !conn || !nq->paths) return -1; struct ll_entry* e = nq->paths->head; while (e) { struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)e; if (path->conn == conn) { queue_remove_data(nq->paths, e); queue_entry_free(e); break; } e = e->next; } int remaining = nq->paths ? queue_entry_count(nq->paths) : 0; if (remaining == 0) { if (nq->paths) { queue_free(nq->paths); nq->paths = NULL; } return 1; } return 0; } // ===== NODEINFO process ===== /* Шлёт пиру ошибку несоответствия типа группы (ERR_GROUP_MISMATCH). */ static void topo_group_send_err_group_mismatch(struct TOPO_GROUP* group, struct ETCP_CONN* conn, uint8_t expected_type, uint8_t received_flags) { if (!group || !conn) return; struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (!peer || !peer->accepted) return; struct TOPOMSG_ERR_GROUP_MISMATCH msg = { .h = topo_header(peer, TOPO_SUBCMD_ERR_GROUP_MISMATCH), .expected_type = expected_type, .received_flags = received_flags }; topo_send_control(conn, &msg, sizeof(msg)); } /* Обрабатывает NODEINFO: проверка подписи/версии, обновление узла, пути, роутинг, форвард. */ int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from, const uint8_t* data, size_t len) { if (!group || !from || len < sizeof(struct TOPOMSG_NODEINFO_PKT)) return -1; struct TOPOMSG_NODEINFO_PKT* pkt = (struct TOPOMSG_NODEINFO_PKT*)data; struct TOPOMSG_NODE* ni = &pkt->node; if (ni->group_id != group->group_id) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "NODEINFO group_id mismatch from %s: expected=%016llx got=%016llx nid=%016llx", from->log_name, (unsigned long long)group->group_id, (unsigned long long)ni->group_id, (unsigned long long)ni->node_id); return -1; } // Проверка соответствия типа группы uint8_t expected_flags = (group->group_type == TOPO_GROUP_TYPE_CHAT) ? 0 : TOPO_FLAG_SEND_SUBNETS; if ((ni->flags & TOPO_FLAG_SEND_SUBNETS) != expected_flags) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "Group type mismatch with %s: expected_flags=0x%02x received_flags=0x%02x (local_type=%u)", from->log_name, expected_flags, ni->flags, group->group_type); topo_group_send_err_group_mismatch(group, from, group->group_type, ni->flags); return -1; } int dyn = topo_node_dyn_size(ni); if ((size_t)dyn + sizeof(struct TOPOMSG_NODEINFO_PKT) != len) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "Incorrect packet size (%s)", from->log_name); return -1; } uint64_t node_id = ni->node_id; if (ni->hop_count >= MAX_HOPS) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "NODEINFO from %s dropped: too many hops (%d)", from->log_name, ni->hop_count); return -1; } if (node_id == 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "NODEINFO node_id=0 from %s grp=%016llx — rejected (malformed)", from->log_name, (unsigned long long)group->group_id); return -1; } if (node_id == group->instance->node_id) return 0; struct TOPO_GROUP_NODE* nodeinfo1 = topo_node_find_by_id(group, node_id); uint8_t new_ver = ni->ver; DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "NODEINFO: node=%016llx ver=%d v4s=%d v4a=%d v6s=%d v6a=%d hops=%d from=%s", (unsigned long long)node_id, new_ver, ni->local_v4_sockets, ni->local_v4_addrs, ni->local_v6_sockets, ni->local_v6_addrs, ni->hop_count, from->log_name); struct ll_queue* paths = NULL; int is_new_node = 0; if (nodeinfo1) { paths = nodeinfo1->paths; } else { is_new_node = 1; } const uint8_t* ser_data = data + sizeof(struct TOPOMSG_HEADER); size_t ser_len = len - sizeof(struct TOPOMSG_HEADER); struct TOPO_NODE* new_ni = NULL; struct TOPO_NODESUBNETS* new_subnets = NULL; uint64_t* new_hop_list = NULL; uint8_t new_hop_count = 0; uint16_t incoming_cumulative_rtt = 0; if (topo_node_deserialize(group, ser_data, ser_len, &new_ni, &new_subnets, &new_hop_list, &new_hop_count, &incoming_cumulative_rtt) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "NODEINFO deserialize failed from %s nid=%016llx", from->log_name, (unsigned long long)node_id); return -1; } { int v4c = topo_list_count((struct _topo_head*)new_ni->v4_addrs); int v6c = topo_list_count((struct _topo_head*)new_ni->v6_addrs); DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "NODEINFO deser: nid=%016llx v4a=%d v6a=%d from=%s", (unsigned long long)node_id, v4c, v6c, from->log_name); } /* Проверяем целую запись до изменения маршрутов и общего реестра. */ struct TOPO_NODE* current = topo_node_registry_find(group->instance->topo_groups, node_id); int conflict = current && current->timestamp == new_ni->timestamp && memcmp(current->x25519_self_sig, new_ni->x25519_self_sig, 64); if (topo_node_verify(new_ni) < 0 || conflict) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "NODEINFO rejected node=%016llx timestamp=%llu conflict=%d", (unsigned long long)node_id, (unsigned long long)new_ni->timestamp, conflict); topo_node_destroy(group->instance->topo_groups, new_ni); struct TOPO_GROUP_NODE rejected = { .subnets = new_subnets }; topo_nodeq_free_group_fields(group->instance->topo_groups, &rejected); u_free(new_hop_list); return -1; } if (nodeinfo1 && new_ni->timestamp <= nodeinfo1->last_timestamp) { DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "NODEINFO unchanged/stale node=%016llx current=%llu incoming=%llu", (unsigned long long)node_id, (unsigned long long)nodeinfo1->last_timestamp, (unsigned long long)new_ni->timestamp); uint64_t hops[MAX_HOPS]; if (new_hop_count) memcpy(hops, new_hop_list, new_hop_count * sizeof(*hops)); hops[new_hop_count] = from->peer_node_id; topo_group_remove_path(nodeinfo1, from); topo_group_add_path(nodeinfo1, from, hops, new_hop_count + 1, incoming_cumulative_rtt); topo_node_destroy(group->instance->topo_groups, new_ni); struct TOPO_GROUP_NODE rejected = { .subnets = new_subnets }; topo_nodeq_free_group_fields(group->instance->topo_groups, &rejected); u_free(new_hop_list); topo_group_peer_progressed(group, from->peer_node_id); return 0; } uint64_t incoming_timestamp = new_ni->timestamp; if (!nodeinfo1) { struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_GROUP_NODE)); if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group node allocation failed"); topo_node_destroy(group->instance->topo_groups, new_ni); struct TOPO_GROUP_NODE rejected = { .subnets = new_subnets }; topo_nodeq_free_group_fields(group->instance->topo_groups, &rejected); u_free(new_hop_list); return -1; } nodeinfo1 = (struct TOPO_GROUP_NODE*)qe; memset((uint8_t*)nodeinfo1 + sizeof(struct ll_entry), 0, sizeof(*nodeinfo1) - sizeof(struct ll_entry)); } struct TOPO_NODE* stored = topo_node_registry_store(group->instance->topo_groups, new_ni); if (!stored) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "node registry update failed"); topo_node_destroy(group->instance->topo_groups, new_ni); struct TOPO_GROUP_NODE rejected = { .subnets = new_subnets }; topo_nodeq_free_group_fields(group->instance->topo_groups, &rejected); if (is_new_node) queue_entry_free(&nodeinfo1->ll); u_free(new_hop_list); return -1; } new_ni = stored; if (!is_new_node) topo_nodeq_free_group_fields(group->instance->topo_groups, nodeinfo1); nodeinfo1->node_id = node_id; nodeinfo1->subnets = new_subnets; if (is_new_node) queue_data_put_with_index(group->nodes, &nodeinfo1->ll); if (is_new_node) DEBUG_INFO(DEBUG_CATEGORY_BGP, "node added to group: node=%016llx grp=%016llx type=%u ch=%s (bgp)", (unsigned long long)node_id, (unsigned long long)group->group_id, group->group_type, group->channel_id); nodeinfo1->paths = paths; nodeinfo1->last_ver = new_ver; nodeinfo1->last_timestamp = incoming_timestamp; nodeinfo1->radio = (ni->flags & TOPO_FLAG_RADIO) ? 1 : 0; uint64_t hop_list[MAX_HOPS]; if (new_hop_count > 0) memcpy(hop_list, new_hop_list, new_hop_count * 8); hop_list[new_hop_count] = from->peer_node_id; uint8_t extended_count = new_hop_count + 1; u_free(new_hop_list); topo_group_remove_path_by_hop(nodeinfo1, from->peer_node_id); topo_group_add_path(nodeinfo1, from, hop_list, extended_count, incoming_cumulative_rtt); nodeinfo1->conn_presence |= NCONN_BGP; nodeinfo1->conn_up |= NCONN_BGP; if (from->peer_node_id == node_id) { nodeinfo1->conn_presence |= NCONN_DIRECT; nodeinfo1->conn_up |= NCONN_DIRECT; } if (group->group_type != TOPO_GROUP_TYPE_CHAT && group->instance->rt) route_insert(group->instance->rt, nodeinfo1); if (node_id != group->instance->node_id) { sqlite3* sdb = group->instance->topo_sqlite_db; if (sdb) { if (topo_node_sqlite_snapshot_put(sdb, new_ni) < 0) DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot persist received NODEINFO node=%016llx", (unsigned long long)node_id); /* узлы в nodes пишет только merkle-путь; BGP лишь дополняет IP для мемберов */ if (topo_node_sqlite_member_exists(sdb, node_id)) topo_node_sqlite_addrs_put(sdb, node_id, new_ni); if (group->group_type == TOPO_GROUP_TYPE_CHAT && group->channel_id[0]) { /* упрощённый плейсхолдер (source=1): pubkeys из BGP, без подписей. не участвует в merkle-хеше, дополняется полной merkle-записью */ topo_node_sqlite_member_placeholder_put(sdb, group->channel_id, node_id, new_ni->public_key, new_ni->ed25519_public_key); } topo_node_sqlite_nodeinfo_updated(sdb, node_id); } } if (group->instance->control_srv) control_server_notify_node_change(group->instance->control_srv, nodeinfo1); DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "node_updated_cb: node=%016llx has_cb=%d ch=%s type=%d", (unsigned long long)node_id, group->instance->topo_groups->node_updated_cb ? 1 : 0, group->channel_id, group->group_type); if (group->instance->topo_groups->node_updated_cb) group->instance->topo_groups->node_updated_cb(group->instance, node_id, ni->public_key, ni->ed25519_public_key); { /* fire BGP node event callbacks */ int ev = is_new_node ? TOPO_NODE_EVENT_NEW : TOPO_NODE_EVENT_UPDATE; struct topo_node_cbk_entry* c = group->node_cbks; while (c) { c->fn(group, node_id, ev, c->arg); c = c->next; } } topo_fire_nodeinfo_cbk(group->instance, group, nodeinfo1); int hop_count = extended_count; struct ll_entry* se = group->senders_list ? group->senders_list->head : NULL; while (se) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data; if (item->conn) { uint64_t id = item->conn->peer_node_id; if (id != node_id) { int found = 0; for (int i = 0; i < hop_count; i++) if (hop_list[i] == id) found = 1; if (found == 0) { topo_group_send_nodeinfo(group, nodeinfo1, item->conn); } } } se = se->next; } int socks_changed = 1; if (!is_new_node) socks_changed = 1; if (node_id != group->instance->node_id && (is_new_node || socks_changed)) { struct TOPO_NODE* tni = topo_node_registry_find(group->instance->topo_groups, nodeinfo1->node_id); int v4a = tni ? topo_list_count((struct _topo_head*)tni->v4_addrs) : 0; int v6a = tni ? topo_list_count((struct _topo_head*)tni->v6_addrs) : 0; if (v4a > 0 || v6a > 0) route_connectivity_probe_node(group->instance, group, nodeinfo1); } /* Создание линков может вызвать транспортные callbacks: структура группы уже обновлена. */ topo_group_peer_progressed(group, from->peer_node_id); node_conn_direct_update_node(group->instance, node_id); return 0; } /* Обрабатывает WITHDRAW: удаляет узел/пути, чистит роутинг, форвардит withdraw. */ int topo_group_process_withdraw(struct TOPO_GROUP* group, struct ETCP_CONN* sender, const uint8_t* data, size_t len) { if (!group || len < sizeof(struct TOPOMSG_WITHDRAW_PKT)) return -1; struct TOPOMSG_WITHDRAW_PKT* wp = (struct TOPOMSG_WITHDRAW_PKT*)data; uint64_t node_id = wp->node_id, wd_source = wp->wd_source; struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(group, node_id); if (!nq) { DEBUG_INFO(DEBUG_CATEGORY_BGP, "node not found"); return 0; } int ret = topo_group_remove_path_by_hop(nq, wd_source); if (ret == 1) { nq->conn_presence = 0; nq->conn_up = 0; topo_fire_nodeinfo_cbk(group->instance, group, nq); if (group->group_type != TOPO_GROUP_TYPE_CHAT && group->instance && group->instance->rt) route_delete(group->instance->rt, nq); if (group->instance->control_srv) control_server_notify_node_removed(group->instance->control_srv, node_id); topo_nodeq_remove_node(group, nq); DEBUG_INFO(DEBUG_CATEGORY_BGP, "node removed from group: node=%016llx grp=%016llx type=%u ch=%s (withdraw)", (unsigned long long)node_id, (unsigned long long)group->group_id, group->group_type, group->channel_id); topo_group_broadcast_withdraw(group, node_id, wd_source, sender); /* BGP не удаляет member-запись: он работает только со своей RAM-таблицей. Удаление из peers_* — только битая подпись при верификации (member_sync). */ { /* fire BGP REMOVE callback */ struct topo_node_cbk_entry* c = group->node_cbks; while (c) { c->fn(group, node_id, TOPO_NODE_EVENT_REMOVE, c->arg); c = c->next; } } } return 0; } /* Сериализует и шлёт NODEINFO узла конкретному пиру. */ static int topo_emit_nodeinfo(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* node, struct ETCP_CONN* conn, uint16_t cumulative_rtt) { if (!group || !node || !conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: invalid arguments"); return -1; } size_t max_sz = sizeof(struct TOPOMSG_NODEINFO_PKT) + TOPO_NODE_WIRE_MAX_SIZE; uint8_t* p = u_malloc(max_sz); if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: allocation failed"); return -1; } struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (!peer || !peer->accepted) { u_free(p); return 0; } /* initial exchange sends the current state after ACCEPT */ struct TOPOMSG_HEADER h = topo_header(peer, TOPO_SUBCMD_NODEINFO); memcpy(p, &h, sizeof(h)); struct TOPO_NODE* sni = topo_node_registry_find(group->instance->topo_groups, node->node_id); if (!sni) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "send_nodeinfo: node %016llx NOT in registry — skip forward to %s grp=%016llx type=%d ch=%s is_local=%d paths=%d nodes_total=%d", (unsigned long long)node->node_id, conn->log_name, (unsigned long long)group->group_id, group->group_type, group->channel_id[0] ? group->channel_id : "-", (node == group->local_node) ? 1 : 0, node->paths ? queue_entry_count(node->paths) : -1, group->nodes ? queue_entry_count(group->nodes) : -1); u_free(p); return -1; } DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "send_nodeinfo: node %016llx ver=%d grp=%016llx to conn=%p name='%s'", (unsigned long long)node->node_id, sni->ver, (unsigned long long)group->group_id, (void*)conn, conn->log_name); uint8_t sflags = (group->group_type == TOPO_GROUP_TYPE_CHAT) ? 0 : TOPO_FLAG_SEND_SUBNETS; int ser_len = topo_node_serialize(sni, node, group->group_id, sflags, p + sizeof(h), max_sz - sizeof(h), cumulative_rtt); if (ser_len < 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: serialize failed for node %016llx", (unsigned long long)node->node_id); u_free(p); return -1; } struct ll_entry* e = queue_entry_new(0); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: entry allocation failed"); u_free(p); return -1; } e->dgram = p; e->len = (size_t)ser_len + sizeof(h); if (etcp_send(conn, e) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: etcp_send FAILED for node %016llx to %s", (unsigned long long)node->node_id, conn->log_name); u_free(p); queue_entry_free(e); return -1; } return 0; } /* NODEINFO сериализуется только при отправке: указатели на узлы не переживают callback. */ int topo_group_send_nodeinfo(struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* node, struct ETCP_CONN* conn) { if (!group || !node || !conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "send_nodeinfo: invalid arguments"); return -1; } struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (!peer || !peer->accepted) return 0; return topo_tx_node(peer, node->node_id); } enum topo_tx_kind { TOPO_TX_CONTROL, TOPO_TX_NODE, TOPO_TX_TABLE }; struct topo_tx_item { struct topo_tx_item* next; enum topo_tx_kind kind; uint64_t node_id; struct ll_entry* packet; uint64_t* nodes; size_t count, pos, sent; }; static void topo_tx_item_free(struct topo_tx_item* item) { if (item->packet) { queue_dgram_free(item->packet); queue_entry_free(item->packet); } u_free(item->nodes); u_free(item); } static void topo_tx_clear(struct TOPO_GROUP_CONN_ITEM* peer) { if (peer->tx_wake) { uasync_call_soon_cancel(peer->group->instance->ua, peer->tx_wake); peer->tx_wake = NULL; } if (peer->conn && peer->conn->send_input_q) queue_waiter_cancel(peer->conn->send_input_q, &peer->tx_waiter); while (peer->tx_head) { struct topo_tx_item* item = peer->tx_head; peer->tx_head = item->next; topo_tx_item_free(item); } peer->tx_tail = NULL; } static void topo_tx_fail(struct TOPO_GROUP_CONN_ITEM* peer) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group sender failed: group=%016llx peer=%016llx epoch=%016llx", (unsigned long long)peer->group->group_id, (unsigned long long)peer->node_id, (unsigned long long)peer->local_epoch); topo_tx_clear(peer); peer->tx_failed = 1; peer->table_sent = 0; topo_recovery_changed(peer->group); topo_group_connect_changed(peer->group); } static void topo_tx_wait(void* arg); static void topo_tx_schedule(struct TOPO_GROUP_CONN_ITEM* peer) { if (!peer->tx_head || peer->tx_failed || peer->tx_wake || peer->tx_waiter.internal || peer->tx_waiter.call_soon_id) return; peer->tx_wake = uasync_call_soon(peer->group->instance->ua, peer, topo_tx_wait); if (!peer->tx_wake) topo_tx_fail(peer); } static int topo_tx_append(struct TOPO_GROUP_CONN_ITEM* peer, struct topo_tx_item* item) { if (!item || peer->tx_failed) { if (item) topo_tx_item_free(item); topo_tx_fail(peer); return -1; } if (peer->tx_tail) peer->tx_tail->next = item; else peer->tx_head = item; peer->tx_tail = item; topo_tx_schedule(peer); return peer->tx_failed ? -1 : 0; } static int topo_tx_control(struct TOPO_GROUP_CONN_ITEM* peer, const void* data, size_t size) { struct topo_tx_item* item = u_calloc(1, sizeof(*item)); if (!item) { topo_tx_fail(peer); return -1; } item->packet = ll_alloc_lldgram(size); if (!item->packet) { u_free(item); topo_tx_fail(peer); return -1; } memcpy(item->packet->dgram, data, size); item->packet->len = size; item->kind = TOPO_TX_CONTROL; return topo_tx_append(peer, item); } static int topo_tx_node(struct TOPO_GROUP_CONN_ITEM* peer, uint64_t node_id) { /* Последнее изменение занимает своё место после уже ожидающих WITHDRAW. * На один узел хранится не более одного несериализованного NODEINFO. */ struct topo_tx_item** link = &peer->tx_head; struct topo_tx_item* prev = NULL; struct topo_tx_item* item = NULL; while (*link) { if ((*link)->kind == TOPO_TX_NODE && (*link)->node_id == node_id) { item = *link; *link = item->next; if (peer->tx_tail == item) peer->tx_tail = prev; item->next = NULL; break; } prev = *link; link = &(*link)->next; } if (!item) item = u_calloc(1, sizeof(*item)); if (!item) { topo_tx_fail(peer); return -1; } item->kind = TOPO_TX_NODE; item->node_id = node_id; return topo_tx_append(peer, item); } static int topo_tx_table(struct TOPO_GROUP_CONN_ITEM* peer) { for (struct topo_tx_item* item = peer->tx_head; item; item = item->next) if (item->kind == TOPO_TX_TABLE) return 0; struct TOPO_GROUP* group = peer->group; struct topo_tx_item* item = u_calloc(1, sizeof(*item)); if (!item) { topo_tx_fail(peer); return -1; } item->kind = TOPO_TX_TABLE; size_t capacity = (size_t)queue_entry_count(group->nodes) + 1; item->nodes = u_malloc(capacity * sizeof(*item->nodes)); if (!item->nodes) { u_free(item); topo_tx_fail(peer); return -1; } item->nodes[item->count++] = group->local_node->node_id; for (struct ll_entry* e = group->nodes->head; e; e = e->next) { struct TOPO_GROUP_NODE* node = (struct TOPO_GROUP_NODE*)e; if (node != group->local_node && topo_group_should_send_to(node, peer->node_id)) item->nodes[item->count++] = node->node_id; } DEBUG_INFO(DEBUG_CATEGORY_BGP, "group table queued: group=%016llx peer=%016llx epoch=%016llx nodes=%zu", (unsigned long long)group->group_id, (unsigned long long)peer->node_id, (unsigned long long)peer->local_epoch, item->count); return topo_tx_append(peer, item); } static int topo_tx_emit_node(struct TOPO_GROUP_CONN_ITEM* peer, uint64_t id) { struct TOPO_GROUP* group = peer->group; struct TOPO_GROUP_NODE* node = id == group->local_node->node_id ? group->local_node : topo_node_find_by_id(group, id); if (!node || (node != group->local_node && !topo_group_should_send_to(node, peer->node_id))) return 0; return topo_emit_nodeinfo(group, node, peer->conn, node == group->local_node ? 0 : topo_get_chain_rtt(node)) < 0 ? -1 : 1; } static void topo_tx_ready(struct ll_queue* queue, void* arg) { struct TOPO_GROUP_CONN_ITEM* peer = arg; /* Deferred waiters must recheck: another producer can occupy the queue first. */ if (queue_entry_count(queue)) { topo_tx_schedule(peer); return; } struct topo_tx_item* item = peer->tx_head; if (!item) return; int result = 0, complete = 1; if (item->kind == TOPO_TX_CONTROL) { result = etcp_send(peer->conn, item->packet); if (result == 0) item->packet = NULL; } else if (item->kind == TOPO_TX_NODE) result = topo_tx_emit_node(peer, item->node_id); else { if (item->pos < item->count) { result = topo_tx_emit_node(peer, item->nodes[item->pos++]); if (result > 0) item->sent++; complete = 0; } else { result = topo_group_send_table_complete(peer->group, peer->conn); if (result == 0) { peer->table_sent = 1; topo_group_log_exchange(peer->group, peer); DEBUG_INFO(DEBUG_CATEGORY_BGP, "group table sent: group=%016llx peer=%016llx epoch=%016llx nodes=%zu", (unsigned long long)peer->group->group_id, (unsigned long long)peer->node_id, (unsigned long long)peer->local_epoch, item->sent); } } } if (result < 0) { topo_tx_fail(peer); return; } topo_group_peer_progressed(peer->group, peer->node_id); if (complete) { peer->tx_head = item->next; if (!peer->tx_head) peer->tx_tail = NULL; topo_tx_item_free(item); } topo_tx_schedule(peer); } static void topo_tx_wait(void* arg) { struct TOPO_GROUP_CONN_ITEM* peer = arg; peer->tx_wake = NULL; if (!peer->conn || !peer->conn->send_input_q || peer->conn->close_requested || queue_waiter_wait(peer->conn->send_input_q, &peer->tx_waiter, topo_tx_ready, peer) < 0) topo_tx_fail(peer); } /* Небольшой ответ вне групповой сессии владеет NCD только до постановки пакета * в транспорт. CLOSED/DOWN отменяет waiter до уничтожения транспортной очереди. */ struct topo_control { struct NODE_CONN_DIRECT* handle; struct ll_queue* queue; struct queue_waiter_handle waiter; struct ll_entry* packet; }; static void topo_control_free(struct topo_control* control) { queue_waiter_cancel(control->queue, &control->waiter); node_conn_direct_close(control->handle); queue_dgram_free(control->packet); queue_entry_free(control->packet); u_free(control); } static void topo_control_event(struct NODE_CONN_DIRECT* handle, enum ncd_event event, void* arg) { (void)handle; if (event == NCD_EVENT_UP) return; DEBUG_WARN(DEBUG_CATEGORY_BGP, "pending group control cancelled transport_event=%d", event); topo_control_free(arg); } static void topo_control_ready(struct ll_queue* queue, void* arg) { struct topo_control* control = arg; if (queue_entry_count(queue)) { if (queue_waiter_wait(queue, &control->waiter, topo_control_ready, control) >= 0) return; DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot rearm group control waiter"); } else if (etcp_send(node_conn_direct_get_conn(control->handle), control->packet) == 0) control->packet = NULL; else DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot send pending group control"); topo_control_free(control); } static int topo_control_wait(struct ETCP_CONN* conn, const void* data, size_t size) { if (!conn->send_input_q || conn->close_requested) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "group control on closing transport"); return -1; } if (!queue_entry_count(conn->send_input_q)) return topo_emit_control(conn, data, size); struct topo_control* control = u_calloc(1, sizeof(*control)); if (!control) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group control allocation failed"); return -1; } control->queue = conn->send_input_q; control->packet = ll_alloc_lldgram(size); if (!control->packet) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group control packet allocation failed"); u_free(control); return -1; } memcpy(control->packet->dgram, data, size); control->packet->len = size; if (node_conn_direct_open(conn->instance, conn->peer_node_id, topo_control_event, control, &control->handle, NULL) == NCD_ERR || queue_waiter_wait(control->queue, &control->waiter, topo_control_ready, control) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot retain pending group control"); topo_control_free(control); return -1; } return 0; } /* Добавляет conn в senders_list (дедуп), если его там ещё нет. */ static int topo_group_add_to_senders(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { if (!group || !conn || !group->senders_list) return -1; struct TOPO_GROUP_CONN_ITEM* pending = topo_group_peer(group, conn->peer_node_id); if (pending) { if (pending->conn && pending->conn != conn) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "group peer has conflicting transport peer=%016llx", (unsigned long long)conn->peer_node_id); return -1; } pending->conn = conn; return 0; } for (struct ll_entry* e = group->senders_list->head; e; e = e->next) if (((struct TOPO_GROUP_CONN_ITEM*)e->data)->conn == conn) return 0; struct ll_entry* e = queue_entry_new(sizeof(struct TOPO_GROUP_CONN_ITEM)); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "senders_add: allocation failed"); return -1; } struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data; item->group = group; item->node_id = conn->peer_node_id; item->conn = conn; item->handle = NULL; item->retained = 1; item->progress = get_time_tb(); if (node_conn_direct_open(conn->instance, conn->peer_node_id, NULL, NULL, &item->handle, NULL) == NCD_ERR) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "senders_add: cannot acquire NCD handle peer=%016llx", (unsigned long long)conn->peer_node_id); queue_entry_free(e); return -1; } queue_data_put(group->senders_list, e); DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "senders_add: acquired peer=%016llx group=%016llx", (unsigned long long)conn->peer_node_id, (unsigned long long)group->group_id); return 0; } /* Решает, надо ли слать узел пиру (не шлём, если пир уже есть в hop_list пути). */ static bool topo_group_should_send_to(const struct TOPO_GROUP_NODE* nq, uint64_t target_id) { if (!nq || !nq->paths) return false; if (target_id == nq->node_id) return false; struct ll_entry* e = nq->paths->head; while (e) { struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)e; uint64_t* hop = (uint64_t*)((uint8_t*)path + sizeof(struct TOPO_NODEPATH)); bool has_id = false; for (uint8_t i = 0; i < path->hop_count; i++) if (hop[i] == target_id) { has_id = true; break; } if (!has_id) return true; e = e->next; } return false; } int topo_group_join_peer(struct TOPO_GROUP* group, struct NODE_CONN_DIRECT* handle) { struct ETCP_CONN* conn = node_conn_direct_get_conn(handle); if (!group || !conn || !conn->links_up) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "join peer requires a connected NCD handle"); return -1; } return topo_group_new_conn(group, conn); } struct topo_leave { struct TOPOMSG_HEADER message; struct UTUN_INSTANCE* instance; uint64_t group_id; struct NODE_CONN_DIRECT* handle; struct ll_queue* queue; struct queue_waiter_handle waiter; }; static void topo_leave_finish(struct topo_leave* leave) { queue_waiter_cancel(leave->queue, &leave->waiter); node_conn_direct_close(leave->handle); u_free(leave); } static void topo_leave_event(struct NODE_CONN_DIRECT* handle, enum ncd_event event, void* arg) { (void)handle; if (event == NCD_EVENT_UP) return; struct topo_leave* leave = arg; DEBUG_WARN(DEBUG_CATEGORY_BGP, "LEAVE_GROUP cancelled: transport event=%d group=%016llx", event, (unsigned long long)leave->group_id); /* NCD CLOSED приходит до уничтожения send_input_q; waiter ещё можно отменить. */ topo_leave_finish(leave); } static void topo_leave_ready(struct ll_queue* queue, void* arg) { (void)queue; struct topo_leave* leave = arg; struct ETCP_CONN* conn = node_conn_direct_get_conn(leave->handle); struct TOPO_GROUP* group = topo_groups_find(leave->instance->topo_groups, leave->group_id); if (group && topo_group_has_sender(group, conn)) { DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "LEAVE_GROUP superseded by a new membership group=%016llx", (unsigned long long)leave->group_id); topo_leave_finish(leave); return; } int result = topo_emit_control(conn, &leave->message, sizeof(leave->message)); if (result != 0) DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot send LEAVE_GROUP group=%016llx", (unsigned long long)leave->group_id); else DEBUG_INFO(DEBUG_CATEGORY_BGP, "LEAVE_GROUP sent: peer=%016llx group=%016llx", (unsigned long long)node_conn_direct_node_id(leave->handle), (unsigned long long)leave->group_id); topo_leave_finish(leave); } int topo_group_leave_peer(struct TOPO_GROUP* group, struct NODE_CONN_DIRECT* handle) { struct ETCP_CONN* conn = node_conn_direct_get_conn(handle); if (!group || !conn) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "leave peer without group/connection"); return -1; } if (!topo_group_has_sender(group, conn)) return 0; if (!conn->send_input_q || conn->close_requested) { topo_group_remove_conn(group, conn, TOPO_REMOVE_LOCAL_LEAVE); return 0; } struct topo_leave* leave = u_calloc(1, sizeof(*leave)); if (!leave) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "LEAVE_GROUP allocation failed"); topo_group_remove_conn(group, conn, TOPO_REMOVE_LOCAL_LEAVE); return -1; } leave->message = topo_header(topo_group_peer(group, conn->peer_node_id), TOPO_SUBCMD_LEAVE_GROUP); leave->instance = group->instance; leave->group_id = group->group_id; leave->queue = conn->send_input_q; if (node_conn_direct_open(group->instance, conn->peer_node_id, topo_leave_event, leave, &leave->handle, NULL) == NCD_ERR) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot retain NCD handle for LEAVE_GROUP"); u_free(leave); topo_group_remove_conn(group, conn, TOPO_REMOVE_LOCAL_LEAVE); return -1; } topo_group_remove_conn(group, conn, TOPO_REMOVE_LOCAL_LEAVE); /* Отправка единственного управляющего пакета после освобождения общей очереди. Контекст не зависит от срока жизни группы; новый JOIN отменяет устаревший LEAVE. */ if (queue_waiter_wait(leave->queue, &leave->waiter, topo_leave_ready, leave) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "cannot wait to send LEAVE_GROUP"); topo_leave_finish(leave); return -1; } return 0; } /* Повторный REQUEST_TABLE той же сессии не создаёт второй снимок. */ static void topo_group_handle_request_table(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (peer && peer->accepted && !peer->table_sent) topo_tx_table(peer); } /* Обрабатывает JOIN_GROUP: для UTUN — добавить запросившего и инициировать BGP обратно. */ static void topo_group_handle_join_group(struct TOPO_GROUP* group, struct ETCP_CONN* conn, const struct TOPOMSG_JOIN_GROUP* msg) { if (topo_group_begin_conn(group, conn, 0) < 0) { topo_group_reject(conn, &msg->h, TOPO_JOIN_UNAVAILABLE); return; } struct TOPO_GROUP_CONN_ITEM* peer = topo_group_peer(group, conn->peer_node_id); if (peer->peer_epoch && peer->peer_epoch != msg->h.src_epoch && peer->accepted) { if (topo_group_reset_exchange(peer) < 0) return; topo_group_send_join_group(group, conn); } peer->peer_epoch = msg->h.src_epoch; struct TOPOMSG_JOIN_GROUP reply = { .h = topo_header(peer, TOPO_SUBCMD_JOIN_ACCEPT), .group_type = group->group_type }; topo_send_control(conn, &reply, sizeof(reply)); } /* Обрабатывает RESYNC: если мы VPN-клиент пира — заново шлём JOIN_GROUP по группам. */ static void topo_group_handle_resync(struct UTUN_INSTANCE* instance, struct ETCP_CONN* conn) { if (!instance || !conn || !conn->instance || !instance->topo_groups) return; /* общий сигнал от пассивной стороны: рестартуем обмен, если мы для этого пира VPN-клиент */ if (!topo_configured(instance, conn->peer_node_id)) return; DEBUG_INFO(DEBUG_CATEGORY_BGP, "handle_resync: from %s — re-announce groups", conn->log_name); struct ll_entry* fe = instance->topo_groups->group_list->head; while (fe) { struct TOPO_GROUP* g = (struct TOPO_GROUP*)fe; fe = fe->next; if (g->group_type != TOPO_GROUP_TYPE_CHAT) topo_group_send_join_group(g, conn); } } /* ── BGP node event callbacks ── */ /* Регистрирует подписчика на события узлов группы (NEW/UPDATE/REMOVE). */ void topo_group_add_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg) { if (!group || !fn) return; struct topo_node_cbk_entry* e = u_malloc(sizeof(*e)); if (!e) return; e->fn = fn; e->arg = arg; e->next = group->node_cbks; group->node_cbks = e; } /* Снимает подписчика событий узлов группы. */ void topo_group_remove_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg) { if (!group || !fn) return; struct topo_node_cbk_entry** p = &group->node_cbks; while (*p) { if ((*p)->fn == fn && (*p)->arg == arg) { struct topo_node_cbk_entry* r = *p; *p = r->next; u_free(r); return; } p = &(*p)->next; } } /* ── Global nodeinfo callbacks (instance-level, any node change) ── */ /* Регистрирует глобальный колбэк изменения nodeinfo (уровень инстанса). */ void utun_add_nodeinfo_cbk(struct UTUN_INSTANCE* instance, nodeinfo_cbk_fn fn, void* arg) { if (!instance || !fn) return; struct nodeinfo_cbk_entry* e = u_malloc(sizeof(*e)); if (!e) return; e->fn = fn; e->arg = arg; e->next = instance->nodeinfo_cbks; instance->nodeinfo_cbks = e; } /* Снимает глобальный колбэк nodeinfo. */ void utun_remove_nodeinfo_cbk(struct UTUN_INSTANCE* instance, nodeinfo_cbk_fn fn, void* arg) { if (!instance || !fn) return; struct nodeinfo_cbk_entry** p = &instance->nodeinfo_cbks; while (*p) { if ((*p)->fn == fn && (*p)->arg == arg) { struct nodeinfo_cbk_entry* r = *p; *p = r->next; u_free(r); return; } p = &(*p)->next; } } /* Вызывает все глобальные nodeinfo-колбэки для узла. */ void topo_fire_nodeinfo_cbk(struct UTUN_INSTANCE* instance, struct TOPO_GROUP* group, struct TOPO_GROUP_NODE* node) { if (!instance || !group || !node) return; struct nodeinfo_cbk_entry* c = instance->nodeinfo_cbks; while (c) { c->fn(group, node, c->arg); c = c->next; } } /* Публичная рассылка WITHDRAW об удалении узла. */ void topo_group_send_withdraw(struct TOPO_GROUP* group, uint64_t node_id) { if (!group) return; DEBUG_INFO(DEBUG_CATEGORY_BGP, "node=%016llx", (unsigned long long)node_id); topo_group_broadcast_withdraw(group, node_id, group->instance->node_id, NULL); }