From 64adb799e94b2bfe1a4114bf640170584f4cbadb Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 06:35:37 +0100 Subject: [PATCH] fix(chat): retain member notification data across callback cancellation --- src/chat/chat_core_priv.h | 1 + src/chat/member_sync.c | 66 ++++++++++++++++++++++++++++----------- src/chat/member_sync.h | 4 ++- 3 files changed, 51 insertions(+), 20 deletions(-) diff --git a/src/chat/chat_core_priv.h b/src/chat/chat_core_priv.h index b4bc10a8..5c0f46ce 100644 --- a/src/chat/chat_core_priv.h +++ b/src/chat/chat_core_priv.h @@ -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; удаление канала ждёт их */ diff --git a/src/chat/member_sync.c b/src/chat/member_sync.c index 2a679637..b7ce339e 100644 --- a/src/chat/member_sync.c +++ b/src/chat/member_sync.c @@ -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); } diff --git a/src/chat/member_sync.h b/src/chat/member_sync.h index e6abca94..cf4fdb97 100644 --- a/src/chat/member_sync.h +++ b/src/chat/member_sync.h @@ -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); /*