jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

etcp: переупорядочить close + защита state==2 во всех API-точках; topo: правильная очистка group nodes и registry; chat/member: новый модуль member_sync; chatgui: обновления транспорта и UI

Browse Source
This commit is contained in:
evgeny committed 2026-08-09 12:39:53 +03:00
1 parent b2e4aadbf0
commit 7f901c01e1
38 files changed
+1275 -540
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/Makefile.am
+4 -4
View File
@@ -70,8 +70,8 @@ utun_CORE_SOURCES = \
chat/chat_event.c \
chat/chat_setting.c \
chat/chat_core.c \
chat/chat_msg.c \
chat/chat_whisper.c \
chat/chat_member.c \
chat/chat_msg.c \ chat/chat_whisper.c \
chat/chat_channel.c \
chat/chat_profile.c \
chat/chat_status.c \
@@ -150,8 +150,8 @@ libutun_a_SOURCES = \
chat/chat_event.c \
chat/chat_setting.c \
chat/chat_core.c \
chat/chat_msg.c \
chat/chat_channel.c \
chat/chat_member.c \
chat/chat_msg.c \ chat/chat_channel.c \
chat/chat_profile.c \
chat/chat_status.c \
chat/chat_sync.c \
src/chat/chat_core.c
+5
View File
@@ -13,6 +13,7 @@
#include "chat_core_priv.h"
#include "chat_event.h"
#include "chat_setting.h"
#include "chat_member.h"
#include "member_sync.h"
#include "../utun_instance.h"
@@ -138,6 +139,9 @@ int chat_core_init(struct UTUN_INSTANCE* inst, const char* db_path) {
g_cc.initialized = 1;
/* register adm_tags change callback → CHAT_EVT_MEMBER_UPDATED */
chat_member_init();
/* create media_files table */
media_index_init(g_cc.db);
@@ -664,6 +668,7 @@ int chat_member_tags_commit(struct chat_member_tags* t) {
uinfo ? uinfo : "", NULL, 0, json, sig, stg);
sqlite3_finalize(st);
if (rc != 0) { DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: tags_commit — member_sync_put failed rc=%d", CC_ID, rc); return -1; }
member_sync_broadcast_one(g_cc.inst, t->ch_id, t->node_id);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "%s: tags_commit ch=%s nid=0x%016llx ver=%d keys=%d json=%s",
CC_ID, t->ch_id, (unsigned long long)t->node_id, ver, t->key_count, json);
src/chat/chat_core.h
+3
View File
@@ -122,6 +122,9 @@ void chat_core_collect_conn_list_trampoline(void* arg);
/* Сбор полных метрик одного соединения (arg = u_malloc uint64_t peer_node_id) */
void chat_core_collect_conn_metrics_trampoline(void* arg);
/* Сбор live-снапшота для панели деталей участника (arg = u_malloc uint64_t node_id) */
void chat_core_collect_member_detail_trampoline(void* arg);
/* ── Headless CLI API ── */
/** Вариант для сборки без UTUN_INSTANCE (использует sqlite3* и node_id) */
src/chat/chat_event.c
+3 -1
View File
@@ -35,8 +35,10 @@ void chat_event_post(int type, const uint8_t* data, int len) {
[16]="NODEINFO_UPDATED", [17]="LINK_STATUS_CHANGED",
[18]="INVITE_RECEIVED", [19]="LOCAL_SOCKETS",
[20]="CONN_LIST", [21]="CONN_METRICS",
[22]="MEMBER_DETAIL", [23]="MEMBER_LIST",
[24]="MEMBER_UPDATED", [25]="MEMBER_REMOVED",
};
const char* n = (type >= 1 && type <= 21) ? names[type] : "?";
const char* n = (type >= 1 && type <= 25) ? names[type] : "?";
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "chat_event: %s(%d) data=%d bytes", n, type, len);
if (data && len > 0) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_DEBUG, " ", data, len);
}
src/chat/chat_event.h
+4
View File
@@ -40,6 +40,10 @@ extern "C" {
#define CHAT_EVT_LOCAL_SOCKETS 19 /* JSON: [{"si":id,"nm":"...","ad":"...","po":port,"pt":"TCP|UDP","st":0|1|2,"lk":n},...] */
#define CHAT_EVT_CONN_LIST 20 /* [count:2][entry:46B]* — список ETCP-соединений */
#define CHAT_EVT_CONN_METRICS 21 /* текст — полный дамп метрик выбранного соединения */
#define CHAT_EVT_MEMBER_DETAIL 22 /* [node_id:8][flags:1][own_tcp_active:2][links...][socks...] — live-снапшот для панели деталей участника */
#define CHAT_EVT_MEMBER_LIST 23 /* [ch_id_len:1][ch_id:var][count:2][member:79B]* — полный список мемберов канала */
#define CHAT_EVT_MEMBER_UPDATED 24 /* [ch_id_len:1][ch_id:var][count:1][member:79B] — обновление одного мембера */
#define CHAT_EVT_MEMBER_REMOVED 25 /* [ch_id_len:1][ch_id:var][node_id:8] — мембер удалён из канала */
typedef void (*chat_event_handler_fn)(int type, const uint8_t* data, int len);
src/chat/chat_member.c
+362
View File
@@ -0,0 +1,362 @@
/*
* chat_member.c — единый код запроса мемберов для GUI
*
* Используется: chatgui (desktop), headless CLI, Android.
* Все три потребителя получают данные через chat_core_get_member_list().
*/
#include "chat_core_priv.h"
#include "chat_member.h"
#include "chat_event.h"
#include "member_sync.h"
#include "../utun_instance.h"
#include "../../lib/mem.h"
#include "../../lib/ll_queue.h"
#include "../../lib/json_flat.h"
#include "../../lib/platform_compat.h"
/* ─── парсинг adm_tags в flags ─── */
static uint8_t parse_adm_tags_flags(const char* adm_tags) {
if (!adm_tags || !adm_tags[0]) return 0;
uint8_t flags = 0;
/* supernode */
char buf[256];
if (json_flat_get(adm_tags, "supernode", buf, sizeof(buf)) == 0
&& strcmp(buf, "yes") == 0) flags |= CHAT_MEMBER_FLAG_SUPERNODE;
/* admin: значение — накопительная строка ed...ed..., последний символ e/d */
if (json_flat_get(adm_tags, "admin", buf, sizeof(buf)) == 0) {
const char* last = buf;
for (const char* p = buf; *p; p++)
if (*p == 'e' || *p == 'd') last = p;
if (*last == 'e') flags |= CHAT_MEMBER_FLAG_ADMIN;
}
/* moder — аналогично admin */
if (json_flat_get(adm_tags, "moder", buf, sizeof(buf)) == 0) {
const char* last = buf;
for (const char* p = buf; *p; p++)
if (*p == 'e' || *p == 'd') last = p;
if (*last == 'e') flags |= CHAT_MEMBER_FLAG_MODER;
}
return flags;
}
/* ─── сериализация (struct → wire) ─── */
void chat_core_serialize_member(const struct chat_member_display* src, uint8_t* dst) {
uint64_t nid = src->node_id;
uint16_t rtt = src->rtt;
memcpy(dst, &nid, 8);
dst[8] = src->online;
dst[9] = src->is_self;
dst[10] = src->flags;
memcpy(dst + 11, &rtt, 2);
dst[13] = src->nat_type;
dst[14] = src->storage;
memcpy(dst + 15, src->name, 64);
}
int chat_core_deserialize_member(const uint8_t* src, struct chat_member_display* dst) {
if (!src || !dst) return -1;
memcpy(&dst->node_id, src, 8);
dst->online = src[8];
dst->is_self = src[9];
dst->flags = src[10];
memcpy(&dst->rtt, src + 11, 2);
dst->nat_type = src[13];
dst->storage = src[14];
memcpy(dst->name, src + 15, 64);
dst->name[63] = '\0';
return 0;
}
/* ─── запрос списка мемберов ─── */
int chat_core_get_member_list(const char* ch_id, uint8_t** out, int* count) {
if (!out || !count) return -1;
*out = NULL; *count = 0;
if (!g_cc.initialized || !ch_id || !g_cc.db || !g_cc.inst) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "chat_member: get_list — not initialized");
return -1;
}
char peers_tbl[80]; peers_table_name(ch_id, peers_tbl, sizeof(peers_tbl));
uint64_t my_id = g_cc.my_node_id;
/* 1. запрашиваем всех мемберов */
sqlite3_stmt* st = NULL;
char sql[512];
snprintf(sql, sizeof(sql),
"SELECT p.node_id, COALESCE(n.online,0), COALESCE(n.name,''),"
" COALESCE(p.local_nick,''), COALESCE(p.adm_tags,''),"
" COALESCE(p.storage,0)"
" FROM \"%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id"
" ORDER BY n.online DESC, p.node_id ASC", peers_tbl);
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "chat_member: get_list — query failed ch=%s err=%s",
ch_id, sqlite3_errmsg(g_cc.db));
return -1;
}
struct chat_member_display* members = NULL;
int cap = 0, cnt = 0;
while (sqlite3_step(st) == SQLITE_ROW) {
if (cnt >= cap) {
cap = cap ? cap * 2 : 32;
struct chat_member_display* tmp = u_realloc(members, (size_t)cap * sizeof(*members));
if (!tmp) { u_free(members); sqlite3_finalize(st); return -1; }
members = tmp;
}
struct chat_member_display* m = &members[cnt];
m->node_id = (uint64_t)sqlite3_column_int64(st, 0);
m->online = (uint8_t)sqlite3_column_int(st, 1);
const char* node_name = (const char*)sqlite3_column_text(st, 2);
const char* local_nick = (const char*)sqlite3_column_text(st, 3);
const char* adm_tags = (const char*)sqlite3_column_text(st, 4);
m->storage = (uint8_t)sqlite3_column_int(st, 5);
m->is_self = (uint8_t)(m->node_id == my_id ? 1 : 0);
m->flags = parse_adm_tags_flags(adm_tags);
m->rtt = 0xFFFF;
m->nat_type = 0xFF;
/* display name: local_nick если задан, иначе node name, иначе short hex */
const char* display = (local_nick && local_nick[0]) ? local_nick :
(node_name && node_name[0]) ? node_name : NULL;
if (display) {
strncpy(m->name, display, sizeof(m->name) - 1);
m->name[sizeof(m->name) - 1] = '\0';
} else {
snprintf(m->name, sizeof(m->name), "0x%04X", (unsigned)(m->node_id & 0xFFFF));
}
cnt++;
}
sqlite3_finalize(st);
if (cnt == 0) {
*out = NULL; *count = 0;
return 0;
}
/* 2. RTT и NAT type из node_addresses (batch) */
char* id_list = u_malloc((size_t)cnt * 21 + 4);
if (!id_list) { u_free(members); return -1; }
id_list[0] = '\0';
for (int i = 0; i < cnt; i++) {
if (i > 0) strcat(id_list, ",");
char nbuf[24]; snprintf(nbuf, sizeof(nbuf), "%llu", (unsigned long long)members[i].node_id);
strcat(id_list, nbuf);
}
snprintf(sql, sizeof(sql),
"SELECT node_id, MIN(rtt), MIN(addr_type) FROM node_addresses"
" WHERE node_id IN (%s) AND rtt > 0 GROUP BY node_id", id_list);
u_free(id_list);
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) == SQLITE_OK) {
while (sqlite3_step(st) == SQLITE_ROW) {
uint64_t nid = (uint64_t)sqlite3_column_int64(st, 0);
int rtt_val = sqlite3_column_int(st, 1);
int at = sqlite3_column_int(st, 2);
for (int i = 0; i < cnt; i++) {
if (members[i].node_id == nid) {
members[i].rtt = (uint16_t)(rtt_val > 0xFFFF ? 0xFFFF : rtt_val);
members[i].nat_type = (uint8_t)at;
break;
}
}
}
sqlite3_finalize(st);
}
/* 3. проверяем connected (активное ETCP-соединение) */
for (int i = 0; i < cnt; i++) {
if (!members[i].online) continue;
if (g_cc.inst->connections
&& queue_find_data_by_index(g_cc.inst->connections, (const uint8_t*)&members[i].node_id))
members[i].online = 2; /* 2 = connected (online + active conn) */
}
/* 4. сериализуем в wire-формат */
size_t wire_sz = (size_t)cnt * CHAT_MEMBER_DISPLAY_SIZE;
uint8_t* wire = u_malloc(wire_sz);
if (!wire) { u_free(members); return -1; }
for (int i = 0; i < cnt; i++)
chat_core_serialize_member(&members[i], wire + i * CHAT_MEMBER_DISPLAY_SIZE);
u_free(members);
*out = wire;
*count = cnt;
return 0;
}
/* ─── один мембер ─── */
int chat_core_get_single_member(const char* ch_id, uint64_t node_id, uint8_t* out) {
if (!out) return -1;
if (!g_cc.initialized || !ch_id || !g_cc.db || !g_cc.inst) return -1;
char peers_tbl[80]; peers_table_name(ch_id, peers_tbl, sizeof(peers_tbl));
uint64_t my_id = g_cc.my_node_id;
sqlite3_stmt* st = NULL;
char sql[384];
snprintf(sql, sizeof(sql),
"SELECT COALESCE(n.online,0), COALESCE(n.name,''),"
" COALESCE(p.local_nick,''), COALESCE(p.adm_tags,''),"
" COALESCE(p.storage,0)"
" FROM \"%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id"
" WHERE p.node_id=?", peers_tbl);
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) != SQLITE_OK) return -1;
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
if (sqlite3_step(st) != SQLITE_ROW) { sqlite3_finalize(st); return -1; }
struct chat_member_display m;
m.node_id = node_id;
m.online = (uint8_t)sqlite3_column_int(st, 0);
const char* node_name = (const char*)sqlite3_column_text(st, 1);
const char* local_nick = (const char*)sqlite3_column_text(st, 2);
const char* adm_tags = (const char*)sqlite3_column_text(st, 3);
m.storage = (uint8_t)sqlite3_column_int(st, 4);
m.is_self = (uint8_t)(node_id == my_id ? 1 : 0);
m.flags = parse_adm_tags_flags(adm_tags);
m.rtt = 0xFFFF;
m.nat_type = 0xFF;
const char* display = (local_nick && local_nick[0]) ? local_nick :
(node_name && node_name[0]) ? node_name : NULL;
if (display) {
strncpy(m.name, display, sizeof(m.name) - 1);
m.name[sizeof(m.name) - 1] = '\0';
} else {
snprintf(m.name, sizeof(m.name), "0x%04X", (unsigned)(node_id & 0xFFFF));
}
sqlite3_finalize(st);
/* RTT / NAT */
snprintf(sql, sizeof(sql),
"SELECT MIN(rtt), MIN(addr_type) FROM node_addresses"
" WHERE node_id=? AND rtt > 0");
if (sqlite3_prepare_v2(g_cc.db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
if (sqlite3_step(st) == SQLITE_ROW) {
int rtt_val = sqlite3_column_int(st, 0);
m.rtt = (uint16_t)(rtt_val > 0xFFFF ? 0xFFFF : rtt_val);
m.nat_type = (uint8_t)sqlite3_column_int(st, 1);
}
sqlite3_finalize(st);
}
/* connected */
if (m.online && g_cc.inst->connections
&& queue_find_data_by_index(g_cc.inst->connections, (const uint8_t*)&node_id))
m.online = 2;
chat_core_serialize_member(&m, out);
return 0;
}
/* ─── трамплины для gui_bridge ─── */
void chat_core_request_member_list_trampoline(void* arg) {
if (!arg || !g_cc.initialized) { u_free(arg); return; }
char ch_id[64]; strncpy(ch_id, (const char*)arg, sizeof(ch_id) - 1); ch_id[sizeof(ch_id) - 1] = '\0';
u_free(arg);
uint8_t* members = NULL;
int count = 0;
if (chat_core_get_member_list(ch_id, &members, &count) != 0 || !members) return;
size_t ch_len = strlen(ch_id);
size_t evt_sz = 1 + ch_len + 2 + (size_t)count * CHAT_MEMBER_DISPLAY_SIZE;
uint8_t* evt = u_malloc(evt_sz);
if (!evt) { u_free(members); return; }
uint16_t u16cnt = (uint16_t)count;
evt[0] = (uint8_t)ch_len;
memcpy(evt + 1, ch_id, ch_len);
memcpy(evt + 1 + ch_len, &u16cnt, 2);
memcpy(evt + 1 + ch_len + 2, members, (size_t)count * CHAT_MEMBER_DISPLAY_SIZE);
chat_event_post(CHAT_EVT_MEMBER_LIST, evt, (int)evt_sz);
u_free(evt);
u_free(members);
}
void chat_core_request_member_rtt_trampoline(void* arg) {
if (!arg || !g_cc.initialized) { u_free(arg); return; }
char ch_id[64]; strncpy(ch_id, (const char*)arg, sizeof(ch_id) - 1); ch_id[sizeof(ch_id) - 1] = '\0';
u_free(arg);
uint8_t* members = NULL;
int count = 0;
if (chat_core_get_member_list(ch_id, &members, &count) != 0 || !members) return;
size_t ch_len = strlen(ch_id);
size_t evt_sz = 1 + ch_len + 1 + CHAT_MEMBER_DISPLAY_SIZE;
uint8_t* evt = u_malloc(evt_sz);
if (!evt) { u_free(members); return; }
/* шлём мемберов по одному (RTT мог измениться у каждого) */
uint16_t rtt_nodata = 0xFFFF;
for (int i = 0; i < count; i++) {
uint16_t rtt_val;
memcpy(&rtt_val, members + i * CHAT_MEMBER_DISPLAY_SIZE + 11, 2);
if (rtt_val == rtt_nodata) continue;
evt[0] = (uint8_t)ch_len;
memcpy(evt + 1, ch_id, ch_len);
evt[1 + ch_len] = 1;
memcpy(evt + 1 + ch_len + 1, members + i * CHAT_MEMBER_DISPLAY_SIZE, CHAT_MEMBER_DISPLAY_SIZE);
chat_event_post(CHAT_EVT_MEMBER_UPDATED, evt, (int)evt_sz);
}
u_free(evt);
u_free(members);
}
/* ─── node_props_changed callback (member_sync → chat_event) ─── */
static void on_adm_tags_changed(uint64_t node_id, const char* adm_tags, void* arg) {
(void)adm_tags; (void)arg;
if (!g_cc.initialized || !g_cc.inst || !g_cc.db) return;
struct chat_core_ctx* cc = &g_cc;
for (int i = 0; i < cc->si_count; i++) {
const char* ch_id = cc->si_ch_id[i];
if (!ch_id) continue;
char peers_tbl[80]; peers_table_name(ch_id, peers_tbl, sizeof(peers_tbl));
char sql[256]; snprintf(sql, sizeof(sql),
"SELECT 1 FROM \"%s\" WHERE node_id=?", peers_tbl);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(cc->db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
if (sqlite3_step(st) == SQLITE_ROW) {
sqlite3_finalize(st);
size_t cl = strlen(ch_id);
uint8_t evt[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, ch_id, cl);
evt[1 + cl] = 1;
if (chat_core_get_single_member(ch_id, node_id,
evt + 1 + cl + 1) == 0)
chat_event_post(CHAT_EVT_MEMBER_UPDATED, evt,
1 + (int)cl + 1 + CHAT_MEMBER_DISPLAY_SIZE);
return;
}
sqlite3_finalize(st);
}
}
}
void chat_member_init(void) {
member_sync_add_props_cbk(on_adm_tags_changed, NULL);
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "chat_member: initialized (adm_tags → MEMBER_UPDATED)");
}
src/chat/chat_member.h
+70
View File
@@ -0,0 +1,70 @@
/*
* chat_member.h — единая структура мембера для отображения в GUI
*
* Используется chatgui (desktop), headless CLI, Android.
* Все три потребителя получают данные через chat_core_get_member_list(),
* а не прямыми SQL-запросами.
*
* Формат на проводе (ручная сериализация, фиксированный размер 79 байт):
* [0-7] node_id uint64_t LE
* [8] online uint8_t
* [9] is_self uint8_t
* [10] flags uint8_t 0x01=supernode 0x02=admin 0x04=moder
* [11-12] rtt uint16_t LE (ms, 0xFFFF=нет данных)
* [13] nat_type uint8_t ADDR_TYPE_*
* [14] storage uint8_t
* [15-78] name char[64] local_nick если задан, иначе node name. \0-padded
*/
#ifndef CHAT_MEMBER_H
#define CHAT_MEMBER_H
#include <stdint.h>
#include <stddef.h>
#ifdef __cplusplus
extern "C" {
#endif
#define CHAT_MEMBER_FLAG_SUPERNODE 0x01
#define CHAT_MEMBER_FLAG_ADMIN 0x02
#define CHAT_MEMBER_FLAG_MODER 0x04
#define CHAT_MEMBER_DISPLAY_SIZE 79
/* Получить всех мемберов канала. out = u_malloc-массив, count = количество. */
int chat_core_get_member_list(const char* ch_id, uint8_t** out, int* count);
/* Получить одного мембера по node_id. out = буфер размером CHAT_MEMBER_DISPLAY_SIZE. */
int chat_core_get_single_member(const char* ch_id, uint64_t node_id, uint8_t* out);
/* Сериализовать из внутренней структуры в wire-формат */
struct chat_member_display {
uint64_t node_id;
uint8_t online;
uint8_t is_self;
uint8_t flags;
uint16_t rtt;
uint8_t nat_type;
uint8_t storage;
char name[64];
};
void chat_core_serialize_member(const struct chat_member_display* src, uint8_t* dst);
int chat_core_deserialize_member(const uint8_t* src, struct chat_member_display* dst);
/* Трамплин для gui_bridge: GUI запрашивает полный список мемберов канала.
* arg = вызов u_strdup(ch_id). Результат: CHAT_EVT_MEMBER_LIST. */
void chat_core_request_member_list_trampoline(void* arg);
/* Трамплин для gui_bridge: GUI запрашивает RTT мемберов канала.
* arg = вызов u_strdup(ch_id). Результат: CHAT_EVT_MEMBER_UPDATED с RTT-обновлениями. */
void chat_core_request_member_rtt_trampoline(void* arg);
/* Инициализация: регистрирует node_props_changed callback для рассылки MEMBER_UPDATED. */
void chat_member_init(void);
#ifdef __cplusplus
}
#endif
#endif /* CHAT_MEMBER_H */
src/chat/chat_profile.c
+6 -3
View File
@@ -6,6 +6,7 @@
#include "chat_core_priv.h"
#include "chat_event.h"
#include "chat_member.h"
#include "../../lib/json_flat.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "member_sync.h"
@@ -73,9 +74,11 @@ void chat_core_update_my_name(const char* name) {
int r2 = member_sync_put(g_cc.inst, ch, myid, g_cc.inst->my_keys.public_key,
g_cc.inst->my_ed25519_pubkey, new_sig, join_ts, NULL, 0, juser, NULL, 0, NULL, NULL, 0);
if (r2 != 0) DEBUG_ERROR(DEBUG_CATEGORY_CHAT_SYNC, "%s: member_sync_put(my_name) FAILED ch=%s rc=%d", CC_ID, ch, r2);
uint8_t evt[65]; uint8_t cl = (uint8_t)strlen(ch);
evt[0] = cl; memcpy(evt + 1, ch, cl);
chat_event_post(CHAT_EVT_MEMBERS_CHANGED, evt, 1 + cl);
else member_sync_broadcast_one(g_cc.inst, ch, myid);
uint8_t mevt[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE]; uint8_t cl = (uint8_t)strlen(ch);
mevt[0] = cl; memcpy(mevt + 1, ch, cl); mevt[1 + cl] = 1;
if (chat_core_get_single_member(ch, myid, mevt + 1 + cl + 1) == 0)
chat_event_post(CHAT_EVT_MEMBER_UPDATED, mevt, 1 + cl + 1 + CHAT_MEMBER_DISPLAY_SIZE);
}
sqlite3_finalize(ps);
}
src/chat/chat_status.c
+139
View File
@@ -14,7 +14,9 @@
#include "../../lib/platform_compat.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/etcp_connections.h"
#include "../transport_layer/stcp_link.h"
#include "../transport_layer/pkt_normalizer.h"
#include <string.h>
static const char* nat_type_str(uint8_t t) {
switch (t) { case 0: return "UNKNOWN"; case 1: return "EIM"; case 2: return "STRICT"; case 3: return "DIRECT"; default: return "?"; }
@@ -288,3 +290,140 @@ void chat_core_collect_conn_metrics_trampoline(void* arg) {
u_free(arg);
collect_conn_metrics(peer_id);
}
/* ── Live-снапшот для панели деталей участника ──
* packet: [node_id:8][flags:1][own_tcp_active:2][link_count:1](link:24B)*[sock_count:1](sock:48B)*
* flags: bit0=conn_present bit1=conn_up(links_up) bit2=conn_initialized
* link: [is_tcp:1][link_state:1][link_status:1][tcp_ready:1][family:1][addr_len:1][addr:16][port:2]
* sock: [sock_id:4][port:2][link_count:4][ifname:38]
* Собирается в worker-потоке — читает живые структуры UTUN_INSTANCE безопасно. */
#define MEMBER_DETAIL_MAX_LINKS 16
#define MEMBER_DETAIL_MAX_SOCKS 16
#define MEMBER_DETAIL_LINK_SIZE 24
#define MEMBER_DETAIL_SOCK_SIZE 48
/* Повторяет Qt: name.lastIndexOf("_v") → name.mid(3, idx - 3) */
static void sock_ifname(const char* name, char* out, size_t out_sz) {
out[0] = '\0';
if (!name || !name[0]) return;
const char* last = NULL;
for (const char* q = name; (q = strstr(q, "_v")) != NULL; q++) last = q;
if (!last) return;
int idx = (int)(last - name);
if (idx <= 3) return;
int len = idx - 3;
if (len >= (int)out_sz) len = (int)out_sz - 1;
memcpy(out, name + 3, len);
out[len] = '\0';
}
static void collect_member_detail(uint64_t node_id) {
if (!g_cc.initialized || !g_cc.inst) return;
struct UTUN_INSTANCE* inst = g_cc.inst;
uint8_t flags = 0;
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
if (conn) {
flags |= 1;
if (conn->links_up) flags |= 2;
if (conn->initialized) flags |= 4;
}
uint16_t own_tcp_active = 0;
if (node_id == inst->node_id) {
if (inst->connections) {
for (struct ll_entry* e = inst->connections->head; e; e = e->next) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
if (ce && ce->conn && ce->conn->initialized) {
for (struct ETCP_LINK* lk = ce->conn->links; lk; lk = lk->next)
if (lk->is_tcp) { own_tcp_active++; break; }
}
}
}
if (inst->tcp_connections) {
for (struct ll_entry* e = inst->tcp_connections->head; e; e = e->next) {
struct tcp_conn_entry* te = (struct tcp_conn_entry*)e->data;
if (te && te->etcp_conn && te->etcp_conn->initialized) own_tcp_active++;
}
}
}
uint8_t link_count = 0;
for (struct ETCP_LINK* lk = conn ? conn->links : NULL; lk && link_count < MEMBER_DETAIL_MAX_LINKS; lk = lk->next)
link_count++;
uint8_t sock_count = 0;
for (struct ETCP_SOCKET* s = inst->etcp_sockets; s && sock_count < MEMBER_DETAIL_MAX_SOCKS; s = s->next)
sock_count++;
for (struct TCP_SOCKET* ts = inst->tcp_sockets; ts && sock_count < MEMBER_DETAIL_MAX_SOCKS; ts = ts->next)
sock_count++;
size_t buf_sz = 12 + (size_t)link_count * MEMBER_DETAIL_LINK_SIZE
+ 1 + (size_t)sock_count * MEMBER_DETAIL_SOCK_SIZE;
uint8_t* buf = (uint8_t*)u_malloc(buf_sz);
if (!buf) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "collect_member_detail: buf alloc failed"); return; }
uint8_t* p = buf;
memcpy(p, &node_id, 8); p += 8;
*p++ = flags;
memcpy(p, &own_tcp_active, 2); p += 2;
*p++ = link_count;
for (struct ETCP_LINK* lk = conn ? conn->links : NULL; lk && link_count; lk = lk->next, link_count--) {
*p++ = lk->is_tcp ? 1 : 0;
*p++ = lk->link_state;
*p++ = lk->link_status;
*p++ = (lk->is_tcp && lk->tcp_link && stcp_link_is_ready(lk->tcp_link)) ? 1 : 0;
uint8_t family = 0, addr_len = 0; uint8_t addr[16] = {0}; uint16_t port = 0;
const struct sockaddr_storage* sa = &lk->remote_addr;
if (sa->ss_family == AF_INET) {
struct sockaddr_in* sin = (struct sockaddr_in*)sa;
family = 4; addr_len = 4; memcpy(addr, &sin->sin_addr, 4); port = ntohs(sin->sin_port);
} else if (sa->ss_family == AF_INET6) {
struct sockaddr_in6* sin6 = (struct sockaddr_in6*)sa;
family = 6; addr_len = 16; memcpy(addr, &sin6->sin6_addr, 16); port = ntohs(sin6->sin6_port);
}
*p++ = family;
*p++ = addr_len;
memcpy(p, addr, 16); p += 16;
memcpy(p, &port, 2); p += 2;
}
*p++ = sock_count;
for (struct ETCP_SOCKET* s = inst->etcp_sockets; s && sock_count; s = s->next, sock_count--) {
uint32_t sid = s->sock_id;
uint16_t sp = 0;
const struct sockaddr_storage* sa = s->interface_addr.ss_family ? &s->interface_addr : &s->local_addr;
if (sa->ss_family == AF_INET) sp = ntohs(((const struct sockaddr_in*)sa)->sin_port);
else if (sa->ss_family == AF_INET6) sp = ntohs(((const struct sockaddr_in6*)sa)->sin6_port);
uint32_t lc = s->links_queue ? (uint32_t)queue_entry_count(s->links_queue) : 0;
char ifname[38] = {0}; sock_ifname(s->name, ifname, sizeof(ifname));
memcpy(p, &sid, 4); p += 4;
memcpy(p, &sp, 2); p += 2;
memcpy(p, &lc, 4); p += 4;
memcpy(p, ifname, 38); p += 38;
}
for (struct TCP_SOCKET* ts = inst->tcp_sockets; ts && sock_count; ts = ts->next, sock_count--) {
uint32_t sid = ts->sock_id;
uint16_t sp = 0;
const struct sockaddr_storage* sa = ts->interface_addr.ss_family ? &ts->interface_addr : &ts->local_addr;
if (sa->ss_family == AF_INET) sp = ntohs(((const struct sockaddr_in*)sa)->sin_port);
else if (sa->ss_family == AF_INET6) sp = ntohs(((const struct sockaddr_in6*)sa)->sin6_port);
char ifname[38] = {0}; sock_ifname(ts->name, ifname, sizeof(ifname));
memcpy(p, &sid, 4); p += 4;
memcpy(p, &sp, 2); p += 2;
uint32_t lc = 0; memcpy(p, &lc, 4); p += 4;
memcpy(p, ifname, 38); p += 38;
}
chat_event_post(CHAT_EVT_MEMBER_DETAIL, buf, (int)(p - buf));
u_free(buf);
}
void chat_core_collect_member_detail_trampoline(void* arg) {
uint64_t node_id;
memcpy(&node_id, arg, sizeof(node_id));
u_free(arg);
collect_member_detail(node_id);
}
src/chat/chat_sync.c
+49 -18
View File
@@ -1,6 +1,7 @@
#include "chat_sync.h"
#include "chat_core.h"
#include "chat_core_priv.h"
#include "chat_member.h"
#include "chat_setting.h"
#include "chat_event.h"
#include "../routing_layer/topo_node_sqlite.h"
@@ -361,12 +362,15 @@ static void cs_on_remote_status_changed(uint64_t peer, int online) {
for (int i = 0; i < g_cs->channel_count; i++) {
for (int j = 0; j < g_cs->channels[i].peer_count; j++) {
if (g_cs->channels[i].peer_ids[j] == peer) {
size_t cl = strlen(g_cs->channels[i].channel_id);
if (cl > 63) cl = 63;
uint8_t evt[65]; evt[0] = (uint8_t)cl;
memcpy(evt + 1, g_cs->channels[i].channel_id, cl);
chat_event_post(CHAT_EVT_MEMBERS_CHANGED, evt, 1 + (int)cl);
cs_post_channel_online(g_cs, g_cs->channels[i].channel_id);
const char* ch_id = g_cs->channels[i].channel_id;
size_t cl = strlen(ch_id);
uint8_t evt[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, ch_id, cl);
evt[1 + cl] = 1;
if (chat_core_get_single_member(ch_id, peer, evt + 1 + cl + 1) == 0)
chat_event_post(CHAT_EVT_MEMBER_UPDATED, evt, 1 + (int)cl + 1 + CHAT_MEMBER_DISPLAY_SIZE);
cs_post_channel_online(g_cs, ch_id);
break;
}
}
@@ -383,12 +387,15 @@ static void cs_on_peer_status_changed(uint64_t peer, int online) {
for (int j = 0; j < g_cs->channels[i].peer_count; j++) {
if (g_cs->channels[i].peer_ids[j] == peer) {
found_in_channel = 1;
size_t cl = strlen(g_cs->channels[i].channel_id);
if (cl > 63) cl = 63;
uint8_t evt[65]; evt[0] = (uint8_t)cl;
memcpy(evt + 1, g_cs->channels[i].channel_id, cl);
chat_event_post(CHAT_EVT_MEMBERS_CHANGED, evt, 1 + (int)cl);
cs_post_channel_online(g_cs, g_cs->channels[i].channel_id);
const char* ch_id = g_cs->channels[i].channel_id;
size_t cl = strlen(ch_id);
uint8_t mevt[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE];
mevt[0] = (uint8_t)cl;
memcpy(mevt + 1, ch_id, cl);
mevt[1 + cl] = 1;
if (chat_core_get_single_member(ch_id, peer, mevt + 1 + cl + 1) == 0)
chat_event_post(CHAT_EVT_MEMBER_UPDATED, mevt, 1 + (int)cl + 1 + CHAT_MEMBER_DISPLAY_SIZE);
cs_post_channel_online(g_cs, ch_id);
/* 3. Рассылка дельты всем synced-соседям */
uint8_t st = (uint8_t)online;
@@ -524,7 +531,8 @@ static void cs_on_conn_down(struct ETCP_CONN* conn, int event, void* arg) { (voi
static void cs_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
switch (status) {
case ETCP_CONN_STATUS_UP: cs_on_conn_up(conn, ETCP_CBK_EVENT_UP, arg); break;
case ETCP_CONN_STATUS_DOWN: cs_on_conn_down(conn, ETCP_CBK_EVENT_DOWN, arg); break;
case ETCP_CONN_STATUS_DOWN:
case ETCP_CONN_STATUS_DELETE: cs_on_conn_down(conn, ETCP_CBK_EVENT_DOWN, arg); break;
default: break;
}
}
@@ -1280,8 +1288,10 @@ static void cs_handle_channel_join(struct chat_sync* cs, uint64_t peer,
/* add to channel cache — always refresh to include new peer */
cs_refresh_channels(cs);
{ uint8_t evt[65]; uint8_t cl=(uint8_t)strlen(ch_id); evt[0]=cl; memcpy(evt+1,ch_id,cl);
chat_event_post(CHAT_EVT_MEMBERS_CHANGED, evt, 1+cl); }
{ uint8_t mevt[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE]; uint8_t cl = (uint8_t)strlen(ch_id);
mevt[0] = cl; memcpy(mevt + 1, ch_id, cl); mevt[1 + cl] = 1;
if (chat_core_get_single_member(ch_id, node_id, mevt + 1 + cl + 1) == 0)
chat_event_post(CHAT_EVT_MEMBER_UPDATED, mevt, 1 + cl + 1 + CHAT_MEMBER_DISPLAY_SIZE); }
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN calling cs_post_channel_online ch=%s", CS_ID, ch_id);
cs_post_channel_online(cs, ch_id);
@@ -1440,7 +1450,22 @@ static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
evt[0] = ch_id_len; memcpy(evt + 1, ch_id, ch_id_len);
chat_event_post(CHAT_EVT_CHANNEL_UPDATED, evt, 1 + ch_id_len);
chat_event_post(CHAT_EVT_MEMBERS_CHANGED, evt, 1 + ch_id_len);
/* full member list for the newly joined channel */
{ uint8_t* members = NULL; int mc = 0;
if (chat_core_get_member_list(ch_id, &members, &mc) == 0 && members) {
size_t esz = 1 + ch_id_len + 2 + (size_t)mc * CHAT_MEMBER_DISPLAY_SIZE;
uint8_t* mevt = u_malloc(esz);
if (mevt) {
uint16_t u16 = (uint16_t)mc;
mevt[0] = ch_id_len; memcpy(mevt + 1, ch_id, ch_id_len);
memcpy(mevt + 1 + ch_id_len, &u16, 2);
memcpy(mevt + 1 + ch_id_len + 2, members, (size_t)mc * CHAT_MEMBER_DISPLAY_SIZE);
chat_event_post(CHAT_EVT_MEMBER_LIST, mevt, (int)esz);
u_free(mevt);
}
u_free(members);
}
}
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: WELCOME calling cs_post_channel_online ch=%s", CS_ID, ch_id);
cs_post_channel_online(cs, ch_id);
@@ -1536,7 +1561,10 @@ static void cs_handle_peer_upsert(struct chat_sync* cs, uint64_t peer,
cs_refresh_channels(cs);
{ uint8_t mevt[65]; uint8_t ml = (uint8_t)strlen(ch_id); mevt[0] = ml; memcpy(mevt + 1, ch_id, ml); chat_event_post(CHAT_EVT_MEMBERS_CHANGED, mevt, 1 + ml); }
{ uint8_t mevt[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE]; uint8_t ml = (uint8_t)strlen(ch_id);
mevt[0] = ml; memcpy(mevt + 1, ch_id, ml); mevt[1 + ml] = 1;
if (chat_core_get_single_member(ch_id, node_id, mevt + 1 + ml + 1) == 0)
chat_event_post(CHAT_EVT_MEMBER_UPDATED, mevt, 1 + ml + 1 + CHAT_MEMBER_DISPLAY_SIZE); }
/* propagate to others (except sender and the subject node) */
cs_propagate(cs, ch_id, peer, pl, len);
@@ -1556,7 +1584,10 @@ static void cs_handle_peer_remove(struct chat_sync* cs, uint64_t peer,
topo_node_sqlite_member_del(db, ch_id, node_id);
cs_refresh_channels(cs);
{ uint8_t mevt[65]; uint8_t ml = (uint8_t)strlen(ch_id); mevt[0] = ml; memcpy(mevt + 1, ch_id, ml); chat_event_post(CHAT_EVT_MEMBERS_CHANGED, mevt, 1 + ml); }
{ uint8_t mevt[1 + 64 + 8]; uint8_t ml = (uint8_t)strlen(ch_id);
mevt[0] = ml; memcpy(mevt + 1, ch_id, ml);
memcpy(mevt + 1 + ml, &node_id, 8);
chat_event_post(CHAT_EVT_MEMBER_REMOVED, mevt, 1 + ml + 8); }
cs_propagate(cs, ch_id, peer, pl, len);
src/chat/db_sync.c
+2 -1
View File
@@ -1258,7 +1258,8 @@ static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
static void db_sync_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
switch (status) {
case ETCP_CONN_STATUS_UP: db_sync_on_conn_up(conn, arg); break;
case ETCP_CONN_STATUS_DOWN: db_sync_on_conn_down(conn, arg); break;
case ETCP_CONN_STATUS_DOWN:
case ETCP_CONN_STATUS_DELETE: db_sync_on_conn_down(conn, arg); break;
default: break;
}
}
src/chat/member_sync.c
+81
View File
@@ -511,6 +511,87 @@ void member_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* c
merkle_sync_cancel(inst, peer, ch_id);
}
void member_sync_broadcast_one(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t member_id) {
if (!inst || !ch_id) return;
sqlite3* db = _db(inst); if (!db) return;
char peers_tbl[128]; _peers_table(ch_id, peers_tbl, sizeof(peers_tbl));
sqlite3_stmt* stmt = NULL;
char sql[512]; snprintf(sql, sizeof(sql),
"SELECT node_id, x25519_pubkey, ed25519_pubkey,"
" join_sig, join_ts, update_sig, update_ts, userinfo, adm_tags, adm_tags_sig"
" FROM \"%s\" WHERE node_id=?", peers_tbl);
if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: broadcast_one — query failed ch=%s nid=0x%016llx", MS_ID, ch_id, (unsigned long long)member_id);
return;
}
sqlite3_bind_int64(stmt, 1, (sqlite3_int64)member_id);
if (sqlite3_step(stmt) != SQLITE_ROW) { sqlite3_finalize(stmt); return; }
uint64_t nid = (uint64_t)sqlite3_column_int64(stmt, 0);
const uint8_t* x25 = (const uint8_t*)sqlite3_column_blob(stmt, 1);
const uint8_t* ed = (const uint8_t*)sqlite3_column_blob(stmt, 2);
const uint8_t* sig = (const uint8_t*)sqlite3_column_blob(stmt, 3);
uint64_t jts = (uint64_t)sqlite3_column_int64(stmt, 4);
const uint8_t* usig = (const uint8_t*)sqlite3_column_blob(stmt, 5);
uint64_t uts = (uint64_t)sqlite3_column_int64(stmt, 6);
const char* nm = (const char*)sqlite3_column_text(stmt, 7);
const char* atags = (const char*)sqlite3_column_text(stmt, 8);
const uint8_t* atsig = (const uint8_t*)sqlite3_column_blob(stmt, 9);
if (!x25 || !ed) { sqlite3_finalize(stmt); return; }
uint8_t nl = nm ? (uint8_t)strnlen(nm, 255) : 0;
uint8_t atl = atags ? (uint8_t)strnlen(atags, 255) : 0;
sqlite3_stmt* as = NULL;
sqlite3_prepare_v2(db,
"SELECT family, socket_id, protocol, address, port FROM node_addresses WHERE node_id=?"
" ORDER BY family, address, port", -1, &as, NULL);
uint8_t addrs[2048]; int addr_off = 0; int addr_count = 0;
if (as) {
sqlite3_bind_int64(as, 1, (sqlite3_int64)nid);
while (sqlite3_step(as) == SQLITE_ROW && addr_off < (int)sizeof(addrs) - 9) {
int fam = sqlite3_column_int(as, 0);
int sid = sqlite3_column_int(as, 1);
int proto = sqlite3_column_int(as, 2);
int ip_sz = fam == 4 ? 4 : 16;
if (addr_off + 3 + ip_sz + 2 > (int)sizeof(addrs)) break;
addrs[addr_off++] = (uint8_t)fam;
addrs[addr_off++] = (uint8_t)sid;
addrs[addr_off++] = (uint8_t)proto;
memcpy(addrs + addr_off, sqlite3_column_blob(as, 3), (size_t)ip_sz);
addr_off += ip_sz;
uint16_t p = (uint16_t)sqlite3_column_int(as, 4);
addrs[addr_off++] = (uint8_t)(p >> 8);
addrs[addr_off++] = (uint8_t)(p & 0xFF);
addr_count++;
}
sqlite3_finalize(as);
}
uint8_t flags = (sig && jts) ? PEERS_FLAG_HAS_JOIN : 0;
uint8_t buf[8192]; size_t off = 0;
uint16_t wcnt = 1; memcpy(buf, &wcnt, 2); off += 2;
memcpy(buf + off, &nid, 8); off += 8;
memcpy(buf + off, x25, 32); off += 32;
memcpy(buf + off, ed, 32); off += 32;
buf[off++] = flags;
if (flags & PEERS_FLAG_HAS_JOIN) { memcpy(buf + off, sig, 64); off += 64; memcpy(buf + off, &jts, 8); off += 8; }
if (usig && uts) { memcpy(buf + off, usig, 64); off += 64; memcpy(buf + off, &uts, 8); off += 8; }
else { memset(buf + off, 0, 72); off += 72; }
buf[off++] = nl; if (nl) { memcpy(buf + off, nm, nl); off += nl; }
buf[off++] = atl; if (atl) { memcpy(buf + off, atags, atl); off += atl; }
if (atsig && sqlite3_column_bytes(stmt, 9) >= 64) { memcpy(buf + off, atsig, 64); }
else { memset(buf + off, 0, 64); }
off += 64;
buf[off++] = (uint8_t)addr_count;
memcpy(buf + off, addrs, (size_t)addr_off); off += (size_t)addr_off;
sqlite3_finalize(stmt);
merkle_sync_broadcast(inst, ch_id, inst->node_id, buf, off);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: broadcast_one ch=%s nid=0x%016llx size=%zu",
MS_ID, ch_id, (unsigned long long)member_id, off);
}
int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
uint64_t member_id, const uint8_t* x25519,
const uint8_t* ed25519,
src/chat/member_sync.h
+4
View File
@@ -111,6 +111,10 @@ int member_sync_del(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t mem
/* Количество мемберов в канале (SELECT COUNT из peers_<ch_id>). */
int member_sync_count(struct UTUN_INSTANCE* inst, const char* ch_id);
/* Сериализовать одного мембера и отправить всем подключённым пирам через merkle_sync_broadcast.
Вызывается после локального изменения данных (adm_tags, имя, адреса). */
void member_sync_broadcast_one(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t member_id);
/*
* Установить онлайн-статус узла (nodes.online = 0/1).
*/
src/chat/merkle_sync.c
+1 -1
View File
@@ -647,7 +647,7 @@ void merkle_sync_broadcast(struct UTUN_INSTANCE* inst, const char* ns,
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: broadcast ns=%s from=%016llx len=%zu",
MS_ID, ns, (unsigned long long)from_peer, len);
for (struct ms_session* s = ms->sessions; s; s = s->next) {
if (s->sess_state != SESS_SYNCED || s->peer == from_peer || strcmp(s->ns, ns) != 0) continue;
if (s->peer == from_peer || strcmp(s->ns, ns) != 0) continue;
_send_broadcast_data(ms, s->peer, ns, data, len);
}
}
src/media_delivery/media_delivery.c
+7 -5
View File
@@ -1164,18 +1164,20 @@ static void md_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
if (!conn) return;
uint64_t node_id = conn->peer_node_id;
if (status == ETCP_CONN_STATUS_DOWN || status == ETCP_CONN_STATUS_DELETE) {
/* remove from served_nodes */
switch (status) {
case ETCP_CONN_STATUS_DOWN:
case ETCP_CONN_STATUS_DELETE:
if (md->served_nodes) {
struct ll_entry* e = queue_find_data_by_index(md->served_nodes, &node_id);
if (e) { queue_remove_data(md->served_nodes, e); queue_entry_free(e); }
}
/* mark super_peer as disconnected */
struct media_super_peer* sp = md_super_peer_find(md, node_id);
{ struct media_super_peer* sp = md_super_peer_find(md, node_id);
if (sp) {
sp->connected = 0; sp->hello_done = 0; sp->inflight_count = 0;
if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; }
}
} }
break;
default: break;
}
}
src/routing_layer/topo_group.c
+18 -3
View File
@@ -258,8 +258,23 @@ static void topo_group_destroy(struct TOPO_GROUP* group) {
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) queue_free(group->nodes);
if (group->local_node) { topo_nodeq_free_group_fields(group->instance->topo_groups, group->local_node); u_free(group->local_node); }
if (group->nodes) {
struct ll_entry* e; int nc = 0;
while ((e = queue_data_get(group->nodes)) != NULL) {
struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e;
if (nq->paths) { queue_free(nq->paths); nq->paths = NULL; }
topo_nodeq_free_group_fields(group->instance->topo_groups, nq);
queue_entry_free(e); 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);
}
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 5 done");
}
@@ -357,7 +372,7 @@ void topo_groups_destroy(struct UTUN_INSTANCE* instance) {
while ((re = queue_data_get(g->node_registry)) != NULL) {
struct TOPO_NODE* node;
memcpy(&node, re->data + 8, sizeof(node));
topo_node_registry_unref(g, node->node_id);
topo_node_destroy(g, node);
queue_entry_free(re); ncount++;
}
queue_free(g->node_registry);
src/transport_layer/etcp.c
+27 -11
View File
@@ -404,7 +404,10 @@ void etcp_connection_close(struct ETCP_CONN* etcp) {
// === PHASE 1: detach from external world ===
etcp->state = 2; // deleted — blocks ref_take and etcp_send immediately
if (etcp->links_up != 0) etcp->links_up = 0;
etcp_fire_conn_status(etcp, ETCP_CONN_STATUS_DELETE);
etcp_cbk_fire(etcp, ETCP_CBK_EVENT_DOWN);
// Cancel active timers
@@ -432,10 +435,6 @@ void etcp_connection_close(struct ETCP_CONN* etcp) {
etcp->conn_queue = NULL;
}
etcp->state = 2; // deleted
etcp_fire_conn_status(etcp, ETCP_CONN_STATUS_DELETE);
// === PHASE 2: deferred resource cleanup (only if no outstanding refs) ===
if (etcp->ref_count == 0)
uasync_call_soon(etcp->instance->ua, etcp, etcp_connection_free_deferred);
@@ -463,6 +462,10 @@ void etcp_conn_ref_free(struct ETCP_CONN* conn) {
// Reset connection
void etcp_conn_reset(struct ETCP_CONN* etcp) {
if (!etcp || etcp->state == 2) {
if (etcp) DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] conn_reset on deleted conn", etcp->log_name);
return;
}
// Reset IDs
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "Resetting ETCP instance [%s]", etcp->log_name);
etcp->next_tx_id = 1;
@@ -589,6 +592,7 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) {
}
void etcp_links_reset(struct ETCP_CONN* etcp) {// Если сбой в обмене или ребутнулась одна из сторон -> необходимо заново переинициализировать соединение
if (!etcp || etcp->state == 2) return;
// Сбрасываем initialized во всех линках
struct ETCP_LINK* link = etcp->links;
while (link) {
@@ -598,6 +602,8 @@ void etcp_links_reset(struct ETCP_CONN* etcp) {// Если сбой в обме
}
void etcp_conn_reinit(struct ETCP_CONN* etcp, const char* reason) {// Если сбой в обмене или ребутнулась одна из сторон -> необходимо заново переинициализировать соединение
if (!etcp) return;
if (etcp->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] conn_reinit on deleted conn", etcp->log_name); return; }
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] REINIT: %s (init=%d links=%d reinit=%u tx=%d)",
etcp->log_name, reason, etcp->initialized, etcp->links_up, etcp->reinit_count, etcp->tx_state);
@@ -632,6 +638,7 @@ void etcp_conn_reinit(struct ETCP_CONN* etcp, const char* reason) {// Если
// внутренняя функция. Вызывается один раз когда первый линк готов.
void etcp_conn_ready(struct ETCP_CONN* conn) {
if (!conn) return;
if (conn->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] conn_ready on deleted conn", conn->log_name); return; }
if (conn->initialized) return; // already ready
conn->initialized = 1;
@@ -718,6 +725,11 @@ int etcp_int_send(struct ETCP_CONN* etcp, const void* data, uint16_t len) {
return -1;
}
if (etcp->state == 2) {
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] int_send on deleted conn", etcp->log_name);
return -1;
}
if (!etcp->input_queue) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] input_queue is NULL for etcp=%p", etcp->log_name, etcp);
return -1;
@@ -801,7 +813,7 @@ static void input_queue_try_resume(struct ETCP_CONN* etcp) {// при ACK
// Called from link-level etcp_link_update_inflight_lim() after inflight_lim_bytes changes.
// Recalculates connection-level optimal_inflight and resumes input_queue if room opened up.
void etcp_conn_on_inflight_lim_changed(struct ETCP_CONN* etcp) {
if (!etcp) return;
if (!etcp || etcp->state == 2) return;
uint32_t sum = 0;
for (struct ETCP_LINK* tl = etcp->links; tl; tl = tl->next) sum += tl->inflight_lim_bytes;
etcp->optimal_inflight = sum;
@@ -991,6 +1003,7 @@ static void send_ack_req_cb(struct ll_queue* q, void* arg) {// etcp->ack_q data
*/
void etcp_on_link_down(struct ETCP_CONN* etcp, struct ETCP_LINK* down_link) {
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
if (!etcp || etcp->state == 2) return;
int up=0;
struct ETCP_LINK* link = etcp->links;
while (link) {
@@ -1070,6 +1083,8 @@ static void etcp_conn_process_send_queue(struct ETCP_CONN* etcp) {// вызыв
// вызывается линком когда освобождается или очередью если появляются данные на передачу
struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp) {
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
if (!etcp) return NULL;
if (etcp->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] request_pkt on deleted conn", etcp->log_name); return NULL; }
struct ETCP_LINK* link = NULL;
// если есть активный burst — используем этот линк напрямую
for (struct ETCP_LINK* l = etcp->links; l; l = l->next) {
@@ -1385,6 +1400,7 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) {
void etcp_ack_recv(struct ETCP_CONN* etcp, uint32_t seq, uint16_t ts, uint16_t dts) {
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
if (!etcp) return;
if (etcp->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] ack_recv on deleted conn", etcp->log_name); return; }
// DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] processing ACK for seq=%u, ts=%u, dts=%u", etcp->log_name, seq, ts, dts);
@@ -1820,7 +1836,7 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
}
void etcp_update_mtu(struct ETCP_CONN* etcp) {
if (!etcp) return;
if (!etcp || etcp->state == 2) return;
int new_mtu = PACKET_DATA_MAX_MTU;
int has_links = 0;
struct ETCP_LINK* link = etcp->links;
@@ -1852,7 +1868,7 @@ void etcp_metrics_init(struct etcp_metrics* m) {
}
void etcp_metrics_add_rtt(struct ETCP_CONN* etcp, uint16_t rtt_tb) {
if (!etcp) return;
if (!etcp || etcp->state == 2) return;
struct etcp_metrics* m = &etcp->metrics;
int i;
for (i = 0; i < (int)ETCP_METRICS_RTT_BOUNDS_COUNT; i++) {
@@ -1864,7 +1880,7 @@ void etcp_metrics_add_rtt(struct ETCP_CONN* etcp, uint16_t rtt_tb) {
}
void etcp_metrics_add_sent(struct ETCP_CONN* etcp, uint32_t len) {
if (!etcp) return;
if (!etcp || etcp->state == 2) return;
struct etcp_metrics* m = &etcp->metrics;
m->work_sent++;
m->work_bytes_sent += len;
@@ -1873,7 +1889,7 @@ void etcp_metrics_add_sent(struct ETCP_CONN* etcp, uint32_t len) {
}
void etcp_metrics_add_rcvd(struct ETCP_CONN* etcp, uint32_t len) {
if (!etcp) return;
if (!etcp || etcp->state == 2) return;
struct etcp_metrics* m = &etcp->metrics;
m->work_rcvd++;
m->work_bytes_rcvd += len;
@@ -1882,7 +1898,7 @@ void etcp_metrics_add_rcvd(struct ETCP_CONN* etcp, uint32_t len) {
}
void etcp_metrics_add_loss(struct ETCP_CONN* etcp, uint32_t count) {
if (!etcp) return;
if (!etcp || etcp->state == 2) return;
struct etcp_metrics* m = &etcp->metrics;
m->work_lost += count;
m->total_lost += count;
@@ -2001,7 +2017,7 @@ static void metrics_snapshot_timer_cb(void* arg) {
}
void etcp_metrics_start_timer(struct ETCP_CONN* etcp) {
if (!etcp || etcp->metrics.timer) return;
if (!etcp || etcp->state == 2 || etcp->metrics.timer) return;
etcp_metrics_init(&etcp->metrics);
etcp->metrics.timer = uasync_set_timeout(etcp->instance->ua, ETCP_METRICS_INTERVAL_TB, etcp, metrics_snapshot_timer_cb, "etcp_metrics");
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] Metrics timer started (interval=%ums)", etcp->log_name, ETCP_METRICS_INTERVAL_TB / 10);
src/transport_layer/etcp_api.c
+1 -1
View File
@@ -149,7 +149,7 @@ void etcp_socket_cbk_fire(struct ETCP_SOCKET* sock, int event) {
while (cbe) { struct etcp_socket_cbk_entry* n = cbe->next; if (cbe->event_mask & event) cbe->fn(sock, event, cbe->arg); cbe = n; }
}
void etcp_set_routing_exchange_state(struct ETCP_CONN* conn, uint8_t new_state) {
if (!conn) return;
if (!conn || conn->state == 2) return;
conn->routing_exchange_active = new_state;
if (new_state >= 3 && conn->bgp_ready_cbk)
conn->bgp_ready_cbk(conn);
src/transport_layer/etcp_connections.c
+2
View File
@@ -909,6 +909,8 @@ void tcp_socket_remove(struct TCP_SOCKET* sock) {
struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn, struct sockaddr_storage* remote_addr, uint8_t is_server) {
DEBUG_TRACE(DEBUG_CATEGORY_CONNECTION, "");
if (!etcp) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "null etcp"); return NULL; }
if (etcp->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTION, "[%s] link_new on deleted conn", etcp->log_name); return NULL; }
int is_tcp = (conn == NULL);
struct ETCP_LINK* link = u_calloc(1, sizeof(struct ETCP_LINK));
src/utun_instance.c
+4 -16
View File
@@ -513,14 +513,7 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
}
queue_free(instance->tcp_connections); instance->tcp_connections = NULL;
while (instance->tcp_sockets) tcp_socket_remove(instance->tcp_sockets);
/* Drain deferred callbacks so stcp_link_close_impl runs before ETCP conn close */
{
int deferred_loops = 0;
while (instance->ua && instance->ua->immediate_queue_head) {
uasync_poll(instance->ua, 0);
if (++deferred_loops > 1000) { DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[DESTROY] J: deferred loop stuck"); break; }
}
}
uasync_drain_immediate(instance->ua);
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[DESTROY] J done — TCP");
/* Phase K: ETCP connections */
@@ -538,14 +531,9 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
queue_free(instance->connections); instance->connections = NULL;
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[DESTROY] K done — connections");
/* Phase L: drain deferred + pings */
{
int deferred_loops = 0;
while (instance->ua && instance->ua->immediate_queue_head) {
uasync_poll(instance->ua, 0);
if (++deferred_loops > 1000) { DEBUG_ERROR(DEBUG_CATEGORY_SYS, "[DESTROY] L: deferred callback loop stuck"); break; }
}
}
uasync_drain_immediate(instance->ua);
/* Phase L: pings */
{
struct PING_CONTEXT* p = instance->pending_pings;
int pc = 0;