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

fix(chat): retain member notification data across callback cancellation

Browse Source
This commit is contained in:
evgeny committed 2026-10-27 06:35:37 +01:00
1 parent 6efe3000d0
commit 64adb799e9
3 files changed
+51 -20
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_core_priv.h
+1
View File
@@ -95,6 +95,7 @@ struct chat_core_ctx {
struct ms_props_cbk* props_cbks; /* was member_sync.c g_props_cbks */
unsigned props_dispatch_depth; /* снятие props-подписки откладывает free до выхода из вложенных callbacks */
struct ms_apply_cbk* apply_cbks; /* was member_sync.c g_apply_cbks */
unsigned apply_dispatch_depth; /* снятые apply callbacks освобождаются после выхода из всех вложенных вызовов */
struct group_avatar* group_avatar;
uint64_t media_cache_bytes; /* was chat_msg.c g_media_cache_bytes */
struct media_submit_ctx* media_submissions; /* исходящие задачи до последнего callback; удаление канала ждёт их */
src/chat/member_sync.c
+47 -19
View File
@@ -315,12 +315,19 @@ static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int afte
struct ms_props_cbk { node_props_changed_fn fn; void* arg; struct ms_props_cbk* next; };
static int member_callback_channel(const char* channel, char out[64]) {
size_t size = channel ? strnlen(channel, 64) : 0;
if (!size || size == 64) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member callback: invalid namespace"); return -1; }
memcpy(out, channel, size + 1); return 0;
}
static void _fire_props_changed(struct UTUN_INSTANCE* inst, uint64_t node_id, const char* adm_tags, const char* channel_id) {
if (!inst || !CC(inst)) return;
char channel[64]; if (member_callback_channel(channel_id, channel)) return;
struct chat_core_ctx* cc = CC(inst);
cc->props_dispatch_depth++;
for (struct ms_props_cbk* pc = cc->props_cbks; pc; pc = pc->next)
if (pc->fn) pc->fn(node_id, adm_tags, channel_id, pc->arg);
if (pc->fn) pc->fn(node_id, adm_tags, channel, pc->arg);
if (--cc->props_dispatch_depth == 0) {
struct ms_props_cbk** p = &cc->props_cbks;
while (*p) {
@@ -362,17 +369,27 @@ static void _fire_apply_cbk(struct UTUN_INSTANCE* inst, const char* ch_id, uint6
const uint8_t* x25519, uint64_t signed_by,
const uint8_t* signature) {
if (!inst || !CC(inst)) return;
struct ms_apply_cbk* ac = CC(inst)->apply_cbks;
while (ac) { ac->fn(ch_id, node_id, x25519, signed_by, signature, ac->arg); ac = ac->next; }
char channel[64]; if (member_callback_channel(ch_id, channel)) return;
struct chat_core_ctx* cc = CC(inst); cc->apply_dispatch_depth++;
for (struct ms_apply_cbk* ac = cc->apply_cbks; ac; ac = ac->next)
if (ac->fn) ac->fn(channel, node_id, x25519, signed_by, signature, ac->arg);
if (--cc->apply_dispatch_depth == 0) {
struct ms_apply_cbk** p = &cc->apply_cbks;
while (*p) {
if (!(*p)->fn) { struct ms_apply_cbk* removed = *p; *p = removed->next; u_free(removed); }
else p = &(*p)->next;
}
}
}
void member_sync_add_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_fn fn, void* arg) {
if (!inst || !fn || !CC(inst)) return;
int member_sync_add_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_fn fn, void* arg) {
if (!inst || !fn || !CC(inst)) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "apply callback: invalid registration"); return -1; }
struct ms_apply_cbk* e = u_malloc(sizeof(*e));
if (!e) return;
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "apply callback: allocation failed"); return -1; }
e->fn = fn; e->arg = arg;
e->next = CC(inst)->apply_cbks;
CC(inst)->apply_cbks = e;
return 0;
}
void member_sync_remove_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_fn fn, void* arg) {
@@ -381,7 +398,8 @@ void member_sync_remove_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_f
while (*p) {
if ((*p)->fn == fn && (*p)->arg == arg) {
struct ms_apply_cbk* r = *p;
*p = r->next; u_free(r);
if (CC(inst)->apply_dispatch_depth) { r->fn = NULL; r->arg = NULL; }
else { *p = r->next; u_free(r); }
return;
}
p = &(*p)->next;
@@ -403,13 +421,18 @@ static void _ms_on_bgp_node(struct TOPO_GROUP* group, uint64_t node_id, int even
char peers_tbl[128]; _peers_table(group->channel_id, peers_tbl, sizeof(peers_tbl));
sqlite3_stmt* st = NULL;
char sql[256]; snprintf(sql, sizeof(sql), "SELECT COALESCE(adm_tags,'') FROM \"%s\" WHERE node_id=?", peers_tbl);
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
if (sqlite3_step(st) == SQLITE_ROW) {
const char* t = (const char*)sqlite3_column_text(st, 0);
if (t) snprintf(tags, sizeof(tags), "%s", t);
}
sqlite3_finalize(st);
int rc = sqlite3_prepare_v2(db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
if (rc == SQLITE_ROW) {
const char* text = (const char*)sqlite3_column_text(st, 0); int bytes = sqlite3_column_bytes(st, 0);
if (!text || sqlite3_column_type(st, 0) != SQLITE_TEXT || bytes > MS_ADM_TAGS_MAX || strlen(text) != (size_t)bytes) rc = SQLITE_CORRUPT;
else memcpy(tags, text, (size_t)bytes + 1);
}
int final = sqlite3_finalize(st);
if ((rc != SQLITE_ROW && rc != SQLITE_DONE) || final != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "member properties read node=%016llx rc=%d finalize=%d: %s",
(unsigned long long)node_id, rc, final, sqlite3_errmsg(db)); return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: bgp_node ev=%d nid=0x%016llx tags_bytes=%zu",
@@ -776,16 +799,21 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id,
}
void member_sync_record_committed(struct UTUN_INSTANCE* inst, const char* ns, uint64_t node_id, unsigned changes) {
merkle_sync_changed(inst, MT_DATA_MEMBERS, ns);
char channel[64]; if (!inst || member_callback_channel(ns, channel)) return;
merkle_sync_changed(inst, MT_DATA_MEMBERS, channel);
if (changes & MS_APPLY_FEEDBACK) chat_feedback_member_committed(inst, node_id);
db_sync_members_changed(inst, ns);
struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, strtoull(ns, NULL, 10)) : NULL;
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
db_sync_members_changed(inst, channel);
uint64_t group_id = strtoull(channel, NULL, 10);
struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) {
_ms_on_bgp_node(g, node_id, 0, inst);
/* Получатель свойств вправе удалить группу/соединение;после callback берём только актуальные handles. */
g = inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT) return;
struct ETCP_CONN* conn = instance_find_conn(inst, node_id);
/* Существующий UP сохраняем; новый транспорт выбирает только автопланировщик. */
if (conn && conn->links_up && !conn->close_requested && topo_group_new_conn(g, conn) < 0)
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: cannot attach UP member channel=%s node=%016llx", MS_ID, ns,
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: cannot attach UP member channel=%s node=%016llx", MS_ID, channel,
(unsigned long long)node_id);
topo_group_connect_changed(g);
}
src/chat/member_sync.h
+3 -1
View File
@@ -106,7 +106,9 @@ int member_sync_validate_pubkey(struct UTUN_INSTANCE* inst, const char* ch_id,
typedef void (*member_apply_cbk_fn)(const char* ch_id, uint64_t node_id,
const uint8_t* x25519, uint64_t signed_by,
const uint8_t* signature, void* arg);
void member_sync_add_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_fn fn, void* arg);
/* Регистрация0/-1;аргументы callback заимствованы только до его возврата. Снятие себя/других подписок
* и вложенный dispatch допустимы;освобождение откладывается. Destroy сервиса — вне callbacks. */
int member_sync_add_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_fn fn, void* arg);
void member_sync_remove_apply_cbk(struct UTUN_INSTANCE* inst, member_apply_cbk_fn fn, void* arg);
/*