From 7c050eb7efeea24114d660c8b358b2a237868db5 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 11:23:44 +0100 Subject: [PATCH] fix(chat): publish complete peer caches and clean pending JOIN state --- src/chat/chat_sync.c | 427 ++++++++++++++------------------ src/chat/chat_sync.h | 19 +- tests/test_chat_sync_schedule.c | 61 ++++- 3 files changed, 259 insertions(+), 248 deletions(-) diff --git a/src/chat/chat_sync.c b/src/chat/chat_sync.c index 48683b4a..baf0ad17 100644 --- a/src/chat/chat_sync.c +++ b/src/chat/chat_sync.c @@ -35,7 +35,6 @@ #include #include -#include #include #ifdef __ANDROID__ #include @@ -81,6 +80,7 @@ struct chat_sync { uint8_t sync_draining; /* новые задания во время flush обслужит следующий проход */ void* invite_send_timer; /* timer waiting for joiner to respond to CHANNEL_INVITE */ uint8_t initialized; + uint8_t direct_bound, admin_bound, join_bound; /* Только успешно установленные service bindings снимаются в destroy. */ struct { uint64_t channel, peer, key, request, attempt; uint8_t inviter, phase; /* 0=idle, 1=connect, 2=info, 3=ready, 4=offer, 5=prepare */ @@ -124,6 +124,18 @@ static void cs_join_release_transport(struct chat_sync* cs) { } } +static void cs_invitation_remove(struct chat_sync* cs, int index) { + for (int i = index + 1; i < cs->pending_inv_in_count; i++) { + cs->pending_inv_in_ch_id[i - 1] = cs->pending_inv_in_ch_id[i]; + cs->pending_inv_in_node_id[i - 1] = cs->pending_inv_in_node_id[i]; + cs->pending_inv_in_key[i - 1] = cs->pending_inv_in_key[i]; + } + cs->pending_inv_in_count--; + cs->pending_inv_in_ch_id[cs->pending_inv_in_count] = 0; + cs->pending_inv_in_node_id[cs->pending_inv_in_count] = 0; + cs->pending_inv_in_key[cs->pending_inv_in_count] = 0; +} + static int cs_join_hold_transport(struct chat_sync* cs, uint64_t node_id, struct TOPO_NODE* ni) { if (cs->join.handle) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join already owns transport"); return -1; } void* timer = uasync_set_timeout(cs->inst->ua, CS_INFO_REQ_TIMEOUT_MS * 10, cs, @@ -185,7 +197,7 @@ static void cs_join_end(struct chat_sync* cs, int result, int notify) { static void cs_join_finish(struct chat_sync* cs, int result) { cs_join_end(cs, result, 1); } static int cs_join_reserve(struct chat_sync* cs, uint64_t channel, uint64_t peer, uint64_t request, int inviter) { - if (!cs || !cs->initialized || !channel || !peer || peer == cs->inst->node_id) { + if (!cs || !cs->initialized || !channel || channel > INT64_MAX || !peer || peer == cs->inst->node_id) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "invalid or stopped local join"); return CHAT_CONNECT_INVALID; } if (cs->join.channel) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "local join busy"); return CHAT_CONNECT_BUSY; } @@ -213,7 +225,7 @@ static void cs_write_group_id(uint8_t* p, const char* ch_id) { static int cs_group_id_to_ch_id(struct UTUN_INSTANCE* inst, uint64_t group_id, char* ch_id, size_t sz) { if (!inst || !inst->topo_groups) return -1; struct TOPO_GROUP* g = topo_groups_find(inst->topo_groups, group_id); - if (!g || !g->channel_id[0]) { + if (!g || g->group_type != TOPO_GROUP_TYPE_CHAT || !g->channel_id[0]) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv — no CHAT group for group=%016llx, drop", CS_ID, (unsigned long long)group_id); return -1; } @@ -256,17 +268,8 @@ static int cs_send_attempt(struct chat_sync* cs, const char* channel, uint64_t p /* ── Channel cache ── */ -static struct channel_cache* cs_find(struct chat_sync* cs, const char* ch_id) { - for (int i = 0; i < cs->channel_count; i++) - if (strcmp(cs->channels[i].channel_id, ch_id) == 0) return &cs->channels[i]; - return NULL; -} - - /* ── Forward declarations for new message handlers ── */ -static void cs_propagate(struct chat_sync* cs, const char* ch_id, uint64_t exclude_id, - const uint8_t* payload, size_t len); static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt); static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, @@ -290,7 +293,7 @@ static void cs_pending_prune(struct chat_sync* cs); /* ── Recv dispatcher ── */ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { - if (!entry || entry->len < 10) { + if (!entry || !entry->dgram || !conn || entry->len < 10) { if (entry) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: recv short entry len=%zu conn=%s", CS_ID, entry->len, conn ? conn->log_name : "?"); if (entry->dgram) u_free(entry->dgram); queue_entry_free(entry); @@ -306,6 +309,10 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { uint64_t group_id; memcpy(&group_id, d + 1, sizeof(group_id)); group_id = be64toh(group_id); + if (!group_id || group_id > INT64_MAX) { + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync: invalid wire channel=%llu", (unsigned long long)group_id); + u_free(entry->dgram); queue_entry_free(entry); return; + } uint8_t type = d[9]; char ch_id[64]; if (cs_group_id_to_ch_id(cs->inst, group_id, ch_id, sizeof(ch_id)) != 0) { @@ -353,8 +360,8 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { } case CS_MSG_CHANNEL_INVITE: cs_handle_channel_invite(cs, peer, ch_id, pl, plen); break; case CS_MSG_SOUND_SETTING: - if (plen >= 1) - db_sync_set_peer_sound(cs->inst, group_id, peer, pl[0] ? 1 : 0); + if (plen == 1 && pl[0] <= 1) db_sync_set_peer_sound(cs->inst, group_id, peer, pl[0]); + else DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "invalid SOUND_SETTING size=%zu", plen); break; default: DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: UNKNOWN msg type=%02x from=%016llx", CS_ID, type, (unsigned long long)peer); break; } @@ -385,27 +392,6 @@ static struct ETCP_CONN* cs_find_conn_for_node(struct UTUN_INSTANCE* inst, uint6 return NULL; } -/* ── helper: find active ETCP_CONN by pubkey (for invite: same physical node, different node_id) ── */ - -static struct ETCP_CONN* cs_find_conn_by_pubkey(struct UTUN_INSTANCE* inst, - const uint8_t* pubkey_bin, - uint64_t* out_node_id) { - struct ll_entry* entry = inst->connections ? inst->connections->head : NULL; - while (entry) { - struct conn_queue_entry* ce = (struct conn_queue_entry*)entry->data; - struct ETCP_CONN* conn = ce->conn; - if (conn && conn->initialized && conn->links_up - && memcmp(conn->crypto_ctx.peer_public_key, pubkey_bin, 32) == 0) { - *out_node_id = conn->peer_node_id; - DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: find_conn_by_pubkey MATCH conn=%s peer=0x%016llx", - CS_ID, conn->log_name, (unsigned long long)conn->peer_node_id); - return conn; - } - entry = entry->next; - } - return NULL; -} - /* ── helper: check if peer has active ETCP link ── */ static int cs_is_peer_online(struct UTUN_INSTANCE* inst, uint64_t peer_id) { @@ -419,6 +405,7 @@ static int cs_is_peer_online(struct UTUN_INSTANCE* inst, uint64_t peer_id) { /* ── post per-channel online peers count to GUI ── */ static void cs_post_channel_online(struct chat_sync* cs, const char* ch_id) { + char channel[64]; snprintf(channel, sizeof(channel), "%s", ch_id); ch_id = channel; int online = 0; struct channel_cache* ch = NULL; for (int i = 0; i < cs->channel_count; i++) @@ -435,7 +422,7 @@ static void cs_post_channel_online(struct chat_sync* cs, const char* ch_id) { if (cl > 63) cl = 63; uint8_t evt[66]; evt[0] = (uint8_t)cl; memcpy(evt + 1, ch_id, cl); - uint16_t oc = (uint16_t)online; memcpy(evt + 1 + cl, &oc, 2); + uint16_t oc = online > UINT16_MAX ? UINT16_MAX : (uint16_t)online; memcpy(evt + 1 + cl, &oc, 2); chat_event_post(cs->inst, CHAT_EVT_CHANNEL_PEERS_ONLINE, evt, 1 + (int)cl + 2); DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: PEERS_ONLINE ch=%.*s total_online=%d", CS_ID, (int)cl, ch_id, online); } @@ -458,8 +445,9 @@ static int cs_sync_member(struct chat_sync* cs, uint64_t channel_id, uint64_t pe if (rc != SQLITE_ROW && rc != SQLITE_DONE) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "sync membership read failed ch=%llu peer=%016llx rc=%d: %s", (unsigned long long)channel_id, (unsigned long long)peer, rc, sqlite3_errmsg(cs->inst->topo_sqlite_db)); - sqlite3_finalize(st); - return found; + int final = sqlite3_finalize(st); + if (final != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "sync membership finalize failed rc=%d", final); + return found && final == SQLITE_OK; } static int cs_sync_peer_available(struct chat_sync* cs, uint64_t peer) { @@ -607,28 +595,28 @@ static void cs_send_all_sound_settings(struct chat_sync* cs, uint64_t peer) { /* ── Peer status change (только GUI + online-бар; онлайн в БД не хранится) ── */ static void cs_on_peer_status_changed(struct chat_sync* cs, uint64_t peer, int online) { - /* 1. GUI + online-бар */ - int found_in_channel = 0; - for (int i = 0; i < cs->channel_count; i++) { - for (int j = 0; j < cs->channels[i].peer_count; j++) { - if (cs->channels[i].peer_ids[j] == peer) { - found_in_channel = 1; - const char* ch_id = 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(cs->inst, ch_id, peer, mevt + 1 + cl + 1) == 0) - chat_event_post(cs->inst, CHAT_EVT_MEMBER_UPDATED, mevt, 1 + (int)cl + 1 + CHAT_MEMBER_DISPLAY_SIZE); - cs_post_channel_online(cs, ch_id); - break; - } - } + int count = 0; + char (*channels)[64] = cs->channel_count ? u_malloc((size_t)cs->channel_count * 64) : NULL; + if (cs->channel_count && !channels) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "peer status snapshot allocation failed"); return; } - - DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: peer_status_changed peer=0x%016llx online=%d channels=%d found=%d", - CS_ID, (unsigned long long)peer, online, cs->channel_count, found_in_channel); + for (int i = 0; i < cs->channel_count; i++) for (int j = 0; j < cs->channels[i].peer_count; j++) { + if (cs->channels[i].peer_ids[j] != peer) continue; + strcpy(channels[count++], cs->channels[i].channel_id); break; + } + for (int i = 0; i < count; i++) { + const char* channel = channels[i]; + if (!topo_groups_find(cs->inst->topo_groups, strtoull(channel, NULL, 10))) continue; + size_t length = strlen(channel); + uint8_t event[1 + 64 + 1 + CHAT_MEMBER_DISPLAY_SIZE]; + event[0] = (uint8_t)length; memcpy(event + 1, channel, length); event[1 + length] = 1; + if (!chat_core_get_single_member(cs->inst, channel, peer, event + 2 + length)) + chat_event_post(cs->inst, CHAT_EVT_MEMBER_UPDATED, event, 2 + (int)length + CHAT_MEMBER_DISPLAY_SIZE); + cs_post_channel_online(cs, channel); + } + u_free(channels); + DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "peer status peer=%016llx online=%d channels=%d", + (unsigned long long)peer, online, count); } static void cs_resume_db_sync(struct chat_sync* cs, const char* ch_id) { @@ -655,29 +643,6 @@ static void _on_member_sync_done(uint64_t peer, const char* ns, int result, void CS_ID, result == MT_OK ? "OK" : "FAIL", ns, (unsigned long long)peer); } -struct invite_sync_arg { - struct chat_sync* cs; - uint64_t node_id; - char ch_id[64]; -}; - -static void _on_invite_sync_done(uint64_t peer, const char* ns, int result, void* arg) { - struct invite_sync_arg* sa = (struct invite_sync_arg*)arg; - DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invite_sync %s ns=%s peer=%016llx sa=%p node=0x%016llx ch=%s", - CS_ID, result == MT_OK ? "OK" : "FAIL", ns, - (unsigned long long)peer, (void*)sa, sa ? (unsigned long long)sa->node_id : 0, sa ? sa->ch_id : "(null)"); - if (result == MT_OK && sa && sa->cs) { - if (sa->node_id == 0 || sa->ch_id[0] == '\0') { - DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invite_sync CORRUPTED sa=%p node_id=0x%016llx ch_id='%s' — skip, will free", - CS_ID, (void*)sa, (unsigned long long)sa->node_id, sa->ch_id); - } else { - cs_post_channel_online(sa->cs, sa->ch_id); - cs_resume_db_sync(sa->cs, sa->ch_id); - } - } - u_free(sa); -} - /* ── send JOIN_INFO_REQ to start join handshake ── */ static int cs_start_channel_join(struct chat_sync* cs, uint64_t peer) { @@ -777,47 +742,68 @@ static void cs_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) { /* ── Periodic refresh from DB ── */ -static void cs_refresh_channels(struct chat_sync* cs) { +static void cs_cache_free(struct channel_cache* channels, int count) { + for (int i = 0; i < count; i++) u_free(channels[i].peer_ids); + u_free(channels); +} + +/* Новый снимок строится целиком без callbacks; ошибка сохраняет предыдущий кеш. */ +static int cs_refresh_channels(struct chat_sync* cs) { cs_pending_prune(cs); - uint8_t buf[4096]; size_t buf_len; - if (chat_core_list_channels(cs->inst, buf, sizeof(buf), &buf_len) != 0) return; - if (buf_len < 2) return; - - uint16_t cnt; memcpy(&cnt, buf, 2); - const uint8_t* p = buf + 2; size_t rem = buf_len - 2; - - for (int i = 0; i < cs->channel_count; i++) - if (cs->channels[i].peer_ids) u_free(cs->channels[i].peer_ids); - if (cs->channels) u_free(cs->channels); - cs->channels = u_calloc(cnt, sizeof(struct channel_cache)); - cs->channel_count = cnt; - if (!cs->channels) { cs->channel_count = 0; return; } - - for (int i = 0; i < (int)cnt && rem >= 1; i++) { - uint8_t id_len = *p++; rem--; - if (rem < id_len) break; - memcpy(cs->channels[i].channel_id, p, id_len); - cs->channels[i].channel_id[id_len] = '\0'; - p += id_len; rem -= id_len; - - uint8_t peer_buf[2048]; size_t peer_len; - if (chat_core_list_peers(cs->inst, cs->channels[i].channel_id, peer_buf, - sizeof(peer_buf), &peer_len) == 0 && peer_len >= 2) { - uint16_t pc; memcpy(&pc, peer_buf, 2); - cs->channels[i].peer_ids = u_calloc(pc, sizeof(uint64_t)); - cs->channels[i].peer_count = pc; - for (uint16_t j = 0; j < pc; j++) - memcpy(&cs->channels[i].peer_ids[j], peer_buf + 2 + j * 8, 8); + struct channel_cache* channels = NULL; int count = 0, capacity = 0; + sqlite3* db = cs->inst->topo_sqlite_db; + sqlite3_stmt* st = NULL; + int rc = sqlite3_prepare_v2(db, "SELECT channel_id FROM channels ORDER BY created_at,channel_id", -1, &st, NULL); + if (rc == SQLITE_OK) while ((rc = sqlite3_step(st)) == SQLITE_ROW) { + const char* id = (const char*)sqlite3_column_text(st, 0); + int bytes = sqlite3_column_bytes(st, 0); + if (!id || bytes <= 0 || bytes >= 64 || strlen(id) != (size_t)bytes || !chat_channel_id(id)) { + rc = SQLITE_CORRUPT; break; } + if (count == capacity) { + int next = capacity ? capacity * 2 : 8; + if (capacity >= 65536) { rc = SQLITE_TOOBIG; break; } + void* grown = u_realloc(channels, (size_t)next * sizeof(*channels)); + if (!grown) { rc = SQLITE_NOMEM; break; } + channels = grown; capacity = next; + } + struct channel_cache* channel = &channels[count++]; memset(channel, 0, sizeof(*channel)); strcpy(channel->channel_id, id); + char sql[160]; snprintf(sql, sizeof(sql), "SELECT node_id FROM peers_%s WHERE deleted=0 ORDER BY node_id", id); + sqlite3_stmt* peers = NULL; int peer_capacity = 0; + int prc = sqlite3_prepare_v2(db, sql, -1, &peers, NULL); + if (prc == SQLITE_OK) while ((prc = sqlite3_step(peers)) == SQLITE_ROW) { + if (sqlite3_column_type(peers, 0) != SQLITE_INTEGER || !sqlite3_column_int64(peers, 0)) { prc = SQLITE_CORRUPT; break; } + if (channel->peer_count == peer_capacity) { + int next = peer_capacity ? peer_capacity * 2 : 16; + if (peer_capacity >= 65536) { prc = SQLITE_TOOBIG; break; } + void* grown = u_realloc(channel->peer_ids, (size_t)next * sizeof(uint64_t)); + if (!grown) { prc = SQLITE_NOMEM; break; } + channel->peer_ids = grown; peer_capacity = next; + } + channel->peer_ids[channel->peer_count++] = (uint64_t)sqlite3_column_int64(peers, 0); + } + int final = sqlite3_finalize(peers); + if (prc != SQLITE_DONE || final != SQLITE_OK) { rc = prc == SQLITE_DONE ? final : prc; break; } } - - for (int i = 0; i < cs->channel_count; i++) - cs_post_channel_online(cs, cs->channels[i].channel_id); + int final = sqlite3_finalize(st); + char (*notifications)[64] = count ? u_malloc((size_t)count * 64) : NULL; + if (rc != SQLITE_DONE || final != SQLITE_OK || (count && !notifications)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "channel cache refresh failed rc=%d finalize=%d channels=%d: %s", + rc, final, count, sqlite3_errmsg(db)); + u_free(notifications); cs_cache_free(channels, count); return -1; + } + for (int i = 0; i < count; i++) strcpy(notifications[i], channels[i].channel_id); + cs_cache_free(cs->channels, cs->channel_count); cs->channels = channels; cs->channel_count = count; + DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "channel cache refreshed channels=%d", count); + /* Получатель может удалить канал или инициировать новый refresh; имена принадлежат этому вызову. */ + for (int i = 0; i < count; i++) cs_post_channel_online(cs, notifications[i]); + u_free(notifications); return 0; } static void refresh_timer_cb(void* arg) { struct chat_sync* cs = (struct chat_sync*)arg; if (!cs || !cs->initialized) return; + cs->refresh_timer = NULL; #ifdef UTUN_HAVE_STANDBY if (standby_get_sleep_tb() > 0) { cs->refresh_wait = standby_wait(cs, refresh_timer_cb); @@ -828,6 +814,7 @@ static void refresh_timer_cb(void* arg) { cs_refresh_channels(cs); cs->refresh_timer = uasync_set_timeout(cs->inst->ua, 30u * 10000u, cs, refresh_timer_cb, "cs_refresh"); + if (!cs->refresh_timer) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "channel cache refresh timer allocation failed"); } /* ── TTL cleanup ── */ @@ -836,11 +823,13 @@ static void refresh_timer_cb(void* arg) { /* ── Public API ── */ int chat_sync_init(struct UTUN_INSTANCE* inst) { - if (!inst) return -1; + if (!inst || !inst->ua || !chat_core_is_initialized(inst)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync: running chat_core required"); return -1; + } if (inst->chat_sync) return 0; DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: init", CS_ID); struct chat_sync* cs = u_calloc(1, sizeof(*cs)); - if (!cs) return -1; + if (!cs) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync context allocation failed"); return -1; } cs->inst = inst; cs->initialized = 1; inst->chat_sync = cs; @@ -863,13 +852,18 @@ int chat_sync_init(struct UTUN_INSTANCE* inst) { member_sync_destroy(inst); chat_join_port_close(cs->port); chat_join_port_release(cs->port); inst->chat_sync = NULL; u_free(cs); return -1; } - etcp_bind(inst, ETCP_ID_CHAT_SYNC, chat_sync_recv_cb); + if (etcp_bind(inst, ETCP_ID_CHAT_SYNC, chat_sync_recv_cb)) goto fail; + cs->direct_bound = 1; etcp_add_conn_status_cbk(inst, cs_on_conn_status, cs); + struct etcp_status_cbk_entry* status = inst->conn_status_cbks; + while (status && (status->fn != cs_on_conn_status || status->arg != cs)) status = status->next; + if (!status) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync status subscription failed"); goto fail; } - cs_refresh_channels(cs); + if (cs_refresh_channels(cs)) goto fail; cs->refresh_timer = uasync_set_timeout(inst->ua, 50000u, cs, refresh_timer_cb, "cs_refresh"); + if (!cs->refresh_timer) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync refresh timer allocation failed"); goto fail; } cs->info_req_timer = NULL; cs->join_timer = NULL; @@ -877,8 +871,13 @@ int chat_sync_init(struct UTUN_INSTANCE* inst) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: member apply subscription failed", CS_ID); chat_sync_destroy(inst); return -1; } utun_add_peer_sleep_cbk(inst, cs_on_peer_sleep, cs); - chat_admin_init(inst); - chat_join_init(inst); + struct peer_sleep_cbk_entry* sleep = inst->peer_sleep_cbks; + while (sleep && (sleep->removed || sleep->fn != cs_on_peer_sleep || sleep->arg != cs)) sleep = sleep->next; + if (!sleep) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync sleep subscription failed"); goto fail; } + if (chat_admin_init(inst)) goto fail; + cs->admin_bound = 1; + if (chat_join_init(inst)) goto fail; + cs->join_bound = 1; /* После restart транспорт может уже быть UP: нового conn_up не будет. */ for (struct ll_entry* e = inst->connections ? inst->connections->head : NULL; e; e = e->next) { @@ -888,6 +887,9 @@ int chat_sync_init(struct UTUN_INSTANCE* inst) { DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: initialized", CS_ID); return 0; +fail: + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "chat_sync initialization failed; releasing partial runtime"); + chat_sync_destroy(inst); return -1; } void chat_sync_quiesce(struct UTUN_INSTANCE* inst) { @@ -914,11 +916,11 @@ void chat_sync_destroy(struct UTUN_INSTANCE* inst) { cs_pending_free_all(cs); group_avatar_destroy(inst); member_sync_destroy(inst); - chat_admin_destroy(inst); - chat_join_destroy(inst); + if (cs->admin_bound) chat_admin_destroy(inst); + if (cs->join_bound) chat_join_destroy(inst); cs->initialized = 0; inst->chat_sync = NULL; - etcp_unbind(inst, ETCP_ID_CHAT_SYNC); + if (cs->direct_bound) etcp_unbind(inst, ETCP_ID_CHAT_SYNC); etcp_remove_conn_status_cbk(inst, cs_on_conn_status, cs); if (cs->refresh_timer) { uasync_cancel_timeout(inst->ua, cs->refresh_timer); cs->refresh_timer = NULL; } #ifdef UTUN_HAVE_STANDBY @@ -927,9 +929,7 @@ void chat_sync_destroy(struct UTUN_INSTANCE* inst) { cs_cancel_proto_timers(cs); cs_join_release_transport(cs); - for (int i = 0; i < cs->channel_count; i++) - if (cs->channels[i].peer_ids) u_free(cs->channels[i].peer_ids); - if (cs->channels) u_free(cs->channels); + cs_cache_free(cs->channels, cs->channel_count); chat_join_port_release(cs->port); u_free(cs); } @@ -953,9 +953,18 @@ void chat_sync_retry_channels_on_socket_change(struct UTUN_INSTANCE* inst) { void chat_sync_on_channel_deleted(struct UTUN_INSTANCE* inst, const char* ch_id) { struct chat_sync* cs = cs_of(inst); - if (!cs || !ch_id || !ch_id[0]) return; - cs_cancel_sync(cs, 0, strtoull(ch_id, NULL, 10), "channel_deleted"); - if (cs->join.channel == strtoull(ch_id, NULL, 10)) cs_join_finish(cs, CHAT_CONNECT_CANCELLED); + uint64_t channel = chat_channel_id(ch_id); + if (!cs || !channel) return; + cs_cancel_sync(cs, 0, channel, "channel_deleted"); + struct join_pending** link = &cs->join_pending; + while (*link) { + struct join_pending* item = *link; + if (strcmp(item->ch_id, ch_id)) { link = &item->next; continue; } + *link = item->next; u_free(item); + } + for (int i = 0; i < cs->pending_inv_in_count; ) { + if (cs->pending_inv_in_ch_id[i] == channel) cs_invitation_remove(cs, i); else i++; + } for (int i = 0; i < cs->channel_count; i++) { if (strcmp(cs->channels[i].channel_id, ch_id) != 0) continue; if (cs->channels[i].peer_ids) u_free(cs->channels[i].peer_ids); @@ -963,8 +972,9 @@ void chat_sync_on_channel_deleted(struct UTUN_INSTANCE* inst, const char* ch_id) (size_t)(cs->channel_count - i - 1) * sizeof(struct channel_cache)); cs->channel_count--; DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: channel removed from cache ch=%s", CS_ID, ch_id); - return; + break; } + if (cs->join.channel == channel) cs_join_finish(cs, CHAT_CONNECT_CANCELLED); } static void cs_node_clear(struct TOPO_NODE* node) { @@ -1036,83 +1046,25 @@ int chat_sync_offer_link(struct UTUN_INSTANCE* inst, uint64_t request, uint64_t /* ─── chat_sync_join_channel: join channel via already-connected peer ─── */ -static void cs_join_existing(struct UTUN_INSTANCE* inst, const char* channel, uint64_t peer, uint64_t key) { +static int cs_join_existing(struct UTUN_INSTANCE* inst, const char* channel, uint64_t peer, uint64_t key) { struct chat_sync* cs = cs_of(inst); - uint64_t group = channel ? strtoull(channel, NULL, 10) : 0; - if (cs_join_reserve(cs, group, peer, 0, 0)) return; + uint64_t group = chat_channel_id(channel); + int rc = cs_join_reserve(cs, group, peer, 0, 0); + if (rc) return rc; cs->join.key = key; - if (cs_join_hold_transport(cs, peer, NULL) < 0 || cs_start_channel_join(cs, peer) < 0) - cs_join_finish(cs, CHAT_CONNECT_ERROR); -} -void chat_sync_join_channel(struct UTUN_INSTANCE* inst, const char* channel, uint64_t peer) { - cs_join_existing(inst, channel, peer, 0); -} - -/* ─── Ed25519 sign / verify helpers ─── */ - -static int cs_ed25519_sign(const uint8_t* privkey, const uint8_t* msg, size_t msg_len, - uint8_t* sig_out) { - EVP_PKEY* pkey = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, privkey, SC_PRIVKEY_SIZE); - if (!pkey) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: EVP_PKEY_new failed", CS_ID); return -1; } - EVP_MD_CTX* mdctx = EVP_MD_CTX_new(); - if (!mdctx) { EVP_PKEY_free(pkey); return -1; } - int ok = (EVP_DigestSignInit(mdctx, NULL, NULL, NULL, pkey) == 1) - && (EVP_DigestSign(mdctx, sig_out, &(size_t){64}, msg, msg_len) == 1); - EVP_MD_CTX_free(mdctx); - EVP_PKEY_free(pkey); - return ok ? 0 : -1; -} - -static int cs_ed25519_verify(const uint8_t* pubkey, const uint8_t* msg, size_t msg_len, - const uint8_t* sig) { - EVP_PKEY* pkey = EVP_PKEY_new_raw_public_key(EVP_PKEY_ED25519, NULL, pubkey, SC_PUBKEY_SIZE); - if (!pkey) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: EVP_PKEY_new pub failed", CS_ID); return -1; } - EVP_MD_CTX* mdctx = EVP_MD_CTX_new(); - if (!mdctx) { EVP_PKEY_free(pkey); return -1; } - int rc = EVP_DigestVerifyInit(mdctx, NULL, NULL, NULL, pkey) - && EVP_DigestVerify(mdctx, sig, 64, msg, msg_len) == 1; - EVP_MD_CTX_free(mdctx); - EVP_PKEY_free(pkey); - return rc ? 0 : -1; -} - -/* ─── Propagation helper ─── */ - -static void cs_propagate(struct chat_sync* cs, const char* ch_id, uint64_t exclude_id, - const uint8_t* payload, size_t len) { - struct channel_cache* ch = cs_find(cs, ch_id); - if (!ch) return; - uint64_t myid = cs->inst->node_id; - for (int i = 0; i < ch->peer_count; i++) { - uint64_t pid = ch->peer_ids[i]; - if (pid == exclude_id || pid == myid) continue; - if (!cs_find_conn_for_node(cs->inst, pid)) continue; - cs_send(cs, ch_id, pid, payload, len); + if (cs_join_hold_transport(cs, peer, NULL) < 0 || cs_start_channel_join(cs, peer) < 0) { + cs_join_finish(cs, CHAT_CONNECT_ERROR); return CHAT_CONNECT_ERROR; } + return 0; } - -static int _get_node_name(sqlite3* db, uint64_t node_id, char* out, size_t sz) { - if (!db || !out || !sz) return -1; - sqlite3_stmt* st = NULL; - sqlite3_prepare_v2(db, "SELECT name FROM nodes WHERE node_id=?", -1, &st, NULL); - if (!st) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: _get_node_name prepare failed node=0x%016llx", CS_ID, (unsigned long long)node_id); return -1; } - sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id); - int rc = -1; - if (sqlite3_step(st) == SQLITE_ROW) { - const unsigned char* nm = sqlite3_column_text(st, 0); - if (nm) { snprintf(out, sz, "%s", nm); rc = 0; } - else { out[0] = '\0'; rc = 0; } - } - sqlite3_finalize(st); - return rc; -} - /* ─── JOIN_INFO_REQ (0x10): joiner → connection-узел ─── */ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, const char* ch_id, const uint8_t* pl, size_t len, uint64_t attempt) { /* payload: [join_key:8][pass_len:1][password] */ - if (len < 9) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_REQ too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; } + if (len < 9 || pl[8] >= INVITE_PASS_MAX || len != 9 + (size_t)pl[8] || memchr(pl + 9, 0, pl[8])) { + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid JOIN_INFO_REQ size=%zu", CS_ID, len); return; + } uint64_t join_key; memcpy(&join_key, pl, 8); uint64_t channel_id = strtoull(ch_id, NULL, 10); DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_REQ ch=%s from=%016llx len=%zu", @@ -1142,11 +1094,8 @@ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, { sqlite3* vdb = cs->inst->topo_sqlite_db; if (topo_node_sqlite_member_get_join(vdb, ch_id, myid, my_join_sig, &my_join_ts) != 0) { - uint8_t join_msg[128]; - my_join_ts = (uint64_t)ntp_time_get_seconds(cs->inst); - int jmlen = member_sync_build_join_msg(ch_x25519, ch_ed, myid, cs->inst->my_keys.public_key, my_join_ts, - join_msg, (int)sizeof(join_msg)); - if (jmlen >= 0) cs_ed25519_sign(cs->inst->my_ed25519_privkey, join_msg, (size_t)jmlen, my_join_sig); + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN_INFO_REQ: own JOIN unavailable ch=%s", ch_id); + uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt); return; } } uint8_t my_update_sig[64] = {0}; uint64_t my_update_ts; @@ -1158,7 +1107,10 @@ static void cs_handle_join_info_req(struct chat_sync* cs, uint64_t peer, } } uint64_t my_signed_by = 0; uint8_t my_signature[64] = {0}; - topo_node_sqlite_member_get_sign(cs->inst->topo_sqlite_db, ch_id, myid, &my_signed_by, my_signature); + if (topo_node_sqlite_member_get_sign(cs->inst->topo_sqlite_db, ch_id, myid, &my_signed_by, my_signature)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN_INFO_REQ: own invitation unavailable ch=%s", ch_id); + uint8_t error = CS_MSG_ERROR; cs_send_attempt(cs, ch_id, peer, &error, 1, attempt); return; + } /* 5. JOIN_INFO_RESP: инфо канала + мой мембер-рекорд */ size_t il = strlen(juser); @@ -1194,7 +1146,10 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, if (1 + cnl + 8 + 32 + 32 + 64 + 8 + 32 + 32 + 64 + 8 + 64 + 8 + 2 + 8 + 64 > len) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_RESP truncated len=%zu", CS_ID, len); return; } - char name[128]; size_t cn = cnl > 127 ? 127 : cnl; memcpy(name, p, cn); name[cn] = '\0'; p += cnl; + if (cnl >= 128 || memchr(p, 0, cnl)) { + DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "JOIN_INFO_RESP: invalid name length=%u", cnl); return; + } + char name[128]; memcpy(name, p, cnl); name[cnl] = '\0'; p += cnl; uint64_t owner; memcpy(&owner, p, 8); p += 8; const uint8_t* ch_x25519 = p; p += 32; const uint8_t* ch_ed = p; p += 32; @@ -1226,14 +1181,15 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, /* verify channel signature */ uint8_t vmsg[CHAT_CHANNEL_DESCRIPTION_MAX]; int vlen = chat_channel_description_data(ch_id, name, owner, ch_x25519, ch_ed, vmsg, sizeof(vmsg)); - if (vlen < 0 || cs_ed25519_verify(ch_ed, vmsg, (size_t)vlen, ch_sig) != 0) { + if (vlen < 0 || sc_ed25519_verify(ch_ed, vmsg, (size_t)vlen, ch_sig) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_INFO_RESP invalid ch_sig ch=%s", CS_ID, ch_id); return; } uint8_t known_x[32], known_ed[32]; - if (topo_node_sqlite_channel_get(cs->inst->topo_sqlite_db, ch_id, NULL, 0, NULL, known_x, known_ed, NULL) == 0 && - (memcmp(known_x, ch_x25519, 32) || memcmp(known_ed, ch_ed, 32))) { + int known = topo_node_sqlite_channel_get(cs->inst->topo_sqlite_db, ch_id, NULL, 0, NULL, known_x, known_ed, NULL); + if (known == TOPO_SQLITE_LOOKUP_ERROR || (known == 0 && + (memcmp(known_x, ch_x25519, 32) || memcmp(known_ed, ch_ed, 32)))) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "join response conflicts with known channel keys"); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } /* Запись описания и участника атомарна; runtime-группа публикуется после commit. */ @@ -1247,7 +1203,7 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, if (!response_conn || c_node_id != peer || memcmp(c_x25519, response_conn->crypto_ctx.peer_public_key, 32)) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "join response connection identity mismatch"); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } - if (sqlite3_exec(db, "SAVEPOINT join_channel", NULL, NULL, NULL) != SQLITE_OK) { + if (!sqlite3_get_autocommit(db) || sqlite3_exec(db, "SAVEPOINT join_channel", NULL, NULL, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join transaction begin: %s", sqlite3_errmsg(db)); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } int mrc = -1; @@ -1258,11 +1214,14 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join channel commit failed member=%d: %s", mrc, sqlite3_errmsg(db)); if (sqlite3_exec(db, "ROLLBACK TO join_channel; RELEASE join_channel", NULL, NULL, NULL) != SQLITE_OK) DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join channel rollback failed: %s", sqlite3_errmsg(db)); + if (!sqlite3_get_autocommit(db)) utun_instance_stop(cs->inst); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } if (cs->info_req_timer) { uasync_cancel_timeout(cs->inst->ua, cs->info_req_timer); cs->info_req_timer = NULL; } + uint64_t active_attempt = cs->join.attempt; if (chat_core_ensure_channel_ready(cs->inst, ch_id) < 0) { cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } member_sync_record_committed(cs->inst, ch_id, c_node_id, (unsigned)mrc); + if (cs->join.attempt != active_attempt || cs->join.phase != 2) return; /* UI мог отменить JOIN/удалить канал. */ DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "join channel committed ch=%s member=%016llx", ch_id, (unsigned long long)c_node_id); /* построить собственный join_sig */ @@ -1274,7 +1233,7 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, join_ts = (uint64_t)ntp_time_get_seconds(cs->inst); int mlen = member_sync_build_join_msg(ch_x25519, ch_ed, myid, my_x25519, join_ts, msg, (int)sizeof(msg)); if (mlen < 0) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join signature input invalid"); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } - if (cs_ed25519_sign(cs->inst->my_ed25519_privkey, msg, (size_t)mlen, join_sig)) { + if (sc_ed25519_sign(cs->inst->my_ed25519_privkey, msg, (size_t)mlen, join_sig) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join signing failed"); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } } @@ -1285,7 +1244,7 @@ static void cs_handle_join_info_resp(struct chat_sync* cs, uint64_t peer, my_update_ts = (uint64_t)ntp_time_get_seconds(cs->inst); uint8_t umsg[MS_USERINFO_MAX + 73]; int ulen = member_sync_build_update_msg(join_sig, my_update_ts, own_info, umsg, (int)sizeof(umsg)); - if (ulen < 0 || cs_ed25519_sign(cs->inst->my_ed25519_privkey, umsg, (size_t)ulen, my_update_sig)) { + if (ulen < 0 || sc_ed25519_sign(cs->inst->my_ed25519_privkey, umsg, (size_t)ulen, my_update_sig) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "join profile signing failed"); cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } } @@ -1466,7 +1425,7 @@ static void cs_handle_join_ready(struct chat_sync* cs, uint64_t peer, if (c) topo_group_connect_node_once(g, c->peer_node_id); } } - chat_core_ensure_channel_ready(cs->inst, ch_id); + if (chat_core_ensure_channel_ready(cs->inst, ch_id)) { cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } cs_refresh_channels(cs); /* Транспорт может быть открыт до добавления участника, без нового UP. */ cs_schedule_sync(cs, strtoull(ch_id, NULL, 10), peer, "join_ready"); @@ -1492,7 +1451,7 @@ static void cs_handle_channel_invite(struct chat_sync* cs, uint64_t peer, /* Same payload format as CHANNEL_INFO_RESP */ const uint8_t* p = pl; uint8_t nl = *p++; - if (nl >= 128 || 1 + nl + 8 + 32 + 32 + 64 + 1 + 64 + 8 + 2 + 8 > (int)len) { + if (nl >= 128 || 1 + nl + 8 + 32 + 32 + 64 + 1 + 64 + 8 + 2 + 8 > (int)len || memchr(p, 0, nl)) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INVITE truncated or name too long", CS_ID); return; } char name[128]; memcpy(name, p, nl); name[nl] = '\0'; p += nl; @@ -1501,6 +1460,7 @@ static void cs_handle_channel_invite(struct chat_sync* cs, uint64_t peer, const uint8_t* ed_pub = p; p += 32; const uint8_t* ch_sig = p; p += 64; uint8_t inv_flags = *p++; + if (inv_flags & ~PEERS_FLAG_HAS_JOIN) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "CHANNEL_INVITE: unknown flags=%u", inv_flags); return; } const uint8_t* inviter_join_sig = NULL; uint64_t inviter_join_ts = 0; if (inv_flags & PEERS_FLAG_HAS_JOIN) { if (p + 72 > pl + len) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INVITE truncated at join_sig/ts from=%016llx", CS_ID, (unsigned long long)peer); return; } @@ -1519,11 +1479,12 @@ static void cs_handle_channel_invite(struct chat_sync* cs, uint64_t peer, uint64_t join_key = 0; if (p + 8 > pl + len) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INVITE truncated (no join_key) from=%016llx", CS_ID, (unsigned long long)peer); return; } memcpy(&join_key, p, 8); p += 8; + if (!join_key) { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "CHANNEL_INVITE: missing join key"); return; } /* Verify channel signature */ uint8_t vmsg[CHAT_CHANNEL_DESCRIPTION_MAX]; int vlen = chat_channel_description_data(ch_id, name, owner, x25519, ed_pub, vmsg, sizeof(vmsg)); - if (vlen < 0 || cs_ed25519_verify(ed_pub, vmsg, (size_t)vlen, ch_sig) != 0) { + if (vlen < 0 || sc_ed25519_verify(ed_pub, vmsg, (size_t)vlen, ch_sig) != SC_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: CHANNEL_INVITE invalid ch_sig ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer); return; } @@ -1563,21 +1524,10 @@ static void cs_handle_channel_invite(struct chat_sync* cs, uint64_t peer, cs->pending_inv_in_ch_id[slot] = ch_num; cs->pending_inv_in_node_id[slot] = peer; cs->pending_inv_in_key[slot] = join_key; - } + } else { DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "CHANNEL_INVITE: pending invitation limit reached"); return; } size_t inv_name_len = 0; char inv_name[128] = ""; - { sqlite3* vdb = cs->inst->topo_sqlite_db; - sqlite3_stmt* st = NULL; - sqlite3_prepare_v2(vdb, "SELECT name FROM nodes WHERE node_id=?", -1, &st, NULL); - if (st) { - sqlite3_bind_int64(st, 1, (sqlite3_int64)peer); - if (sqlite3_step(st) == SQLITE_ROW) { - const unsigned char* nm = sqlite3_column_text(st, 0); - if (nm) { snprintf(inv_name, sizeof(inv_name), "%s", nm); inv_name_len = strlen(inv_name); } - } - sqlite3_finalize(st); - } - } + chat_core_get_node_name(cs->inst, peer, inv_name, sizeof(inv_name)); inv_name_len = strlen(inv_name); uint8_t evt[512]; int epos = 0; uint8_t cl = (uint8_t)strlen(ch_id); evt[epos++] = cl; memcpy(evt + epos, ch_id, cl); epos += cl; evt[epos++] = nl; memcpy(evt + epos, name, nl); epos += nl; @@ -1645,7 +1595,7 @@ static int cs_send_channel_invite(struct chat_sync* cs, struct UTUN_INSTANCE* in void chat_sync_invite_to_channel(struct UTUN_INSTANCE* inst, const char* channel, uint64_t peer) { struct chat_sync* cs = cs_of(inst); - uint64_t group = channel ? strtoull(channel, NULL, 10) : 0; + uint64_t group = chat_channel_id(channel); if (cs_join_reserve(cs, group, peer, 0, 1)) return; if (cs_join_hold_transport(cs, peer, NULL) < 0) { cs_join_finish(cs, CHAT_CONNECT_ERROR); return; } struct ETCP_CONN* conn = node_conn_direct_get_conn(cs->join.handle); @@ -1659,16 +1609,17 @@ void chat_sync_accept_invite(struct UTUN_INSTANCE* inst, uint64_t channel_id, ui struct chat_sync* cs = cs_of(inst); if (!cs || !inst) return; char ch_id[64]; snprintf(ch_id, sizeof(ch_id), "%llu", (unsigned long long)channel_id); - if (channel_id == 0 || inviter_node_id == 0) { + if (channel_id == 0 || channel_id > INT64_MAX || inviter_node_id == 0) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: accept_invite invalid params ch=%llu node=%016llx", CS_ID, (unsigned long long)channel_id, (unsigned long long)inviter_node_id); return; } /* достать join-ключ, сохранённый при приёме CHANNEL_INVITE (ask) */ - uint64_t join_key = 0; + uint64_t join_key = 0; int slot = -1; for (int i = 0; i < cs->pending_inv_in_count; i++) { if (cs->pending_inv_in_ch_id[i] == channel_id && cs->pending_inv_in_node_id[i] == inviter_node_id) { join_key = cs->pending_inv_in_key[i]; + slot = i; break; } } @@ -1679,13 +1630,17 @@ void chat_sync_accept_invite(struct UTUN_INSTANCE* inst, uint64_t channel_id, ui } DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: accept_invite ch=%llu inviter=0x%016llx", CS_ID, (unsigned long long)channel_id, (unsigned long long)inviter_node_id); - cs_join_existing(inst, ch_id, inviter_node_id, join_key); + if (!cs_join_existing(inst, ch_id, inviter_node_id, join_key)) cs_invitation_remove(cs, slot); } void chat_sync_deny_invite(struct UTUN_INSTANCE* inst, uint64_t channel_id, uint64_t inviter_node_id) { - (void)inst; - if (!cs_of(inst)) return; + struct chat_sync* cs = cs_of(inst); + if (!cs) return; + for (int i = 0; i < cs->pending_inv_in_count; ) { + if (cs->pending_inv_in_ch_id[i] == channel_id && cs->pending_inv_in_node_id[i] == inviter_node_id) + cs_invitation_remove(cs, i); + else i++; + } DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: deny_invite ch=%llu inviter=0x%016llx", CS_ID, (unsigned long long)channel_id, (unsigned long long)inviter_node_id); - /* silently ignore — no response needed */ } diff --git a/src/chat/chat_sync.h b/src/chat/chat_sync.h index daff35a7..0661ee22 100644 --- a/src/chat/chat_sync.h +++ b/src/chat/chat_sync.h @@ -7,6 +7,8 @@ * JOIN_INFO_REQ → JOIN_INFO_RESP → JOIN_REQUEST → JOIN_READY; правила — в chat_join.h. * Жизненным циклом управляет chat_service_start/stop. Вызовы — в потоке uasync. * Вход из GUI/JNI через отдельный chat_join_port, который копирует ссылку и удерживает очередь до close. + * Кеш строится полным снимком БД (до65536 каналов/пиров на канал);ошибка сохраняет прежний снимок. + * Callback UI может удалить канал: обходы уведомлений используют собственные копии ID. */ #ifndef CHAT_SYNC_H @@ -41,7 +43,8 @@ struct UTUN_INSTANCE; /* ── Public API ── */ -/* Создать контекст, обработчики join и member_sync. 0/-1; повтор допустим. */ +/* Создать контекст,обработчики/подписки и первый кеш поверх готового chat_core. 0/-1;повтор допустим. + * Ошибка обязательной подписки/таймера/первого кеша откатывает runtime;данные БД сохраняются. */ int chat_sync_init(struct UTUN_INSTANCE* inst); /* uasync: закрыть приём команд и отменить локальный JOIN до остановки workers. */ void chat_sync_quiesce(struct UTUN_INSTANCE* inst); @@ -70,25 +73,19 @@ int chat_sync_offer_link(struct UTUN_INSTANCE* inst, uint64_t request, uint64_t Отправляет CS_MSG_CHANNEL_INVITE с описанием канала, подписями/профилем инвайтера и join_key. */ void chat_sync_invite_to_channel(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t target_node_id); -/* Принять входящее приглашение — продолжить протокол из chat_join.h. */ +/* Принять сохранённое приглашение (до4 пар канал/инвайтер). Успешный запуск убирает его из очереди; + * BUSY/отказ запуска сохраняет предложение. Описание/участник проверяются и сохраняются только в JOIN_INFO_RESP. */ void chat_sync_accept_invite(struct UTUN_INSTANCE* inst, uint64_t channel_id, uint64_t inviter_node_id); -/* Отклонить входящее приглашение */ +/* Убрать пару из очереди предложений без сетевого ответа;освободившийся слот можно использовать повторно. */ void chat_sync_deny_invite(struct UTUN_INSTANCE* inst, uint64_t channel_id, uint64_t inviter_node_id); -/* Присоединиться к каналу через уже подключённый узел (uasync-поток). - Соединение с target_node_id должно быть установлено (links_up && initialized). - Протокол: JOIN_INFO_REQ → JOIN_INFO_RESP → JOIN_REQUEST → JOIN_READY (chat_join.h). - Канал (DB + TOPO_GROUP) создаётся локально при получении JOIN_INFO_RESP; - это само по себе не даёт членства на других узлах. */ -void chat_sync_join_channel(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t target_node_id); - /* Снять паузу и сбросить задержки повторных попыток во всех кешированных каналах с автоматическим планировщиком. Планировщик пересматривает подключения; работающие соединения сохраняются. Вызывается из auto_socket при появлении/изменении сетевой связности. */ void chat_sync_retry_channels_on_socket_change(struct UTUN_INSTANCE* inst); -/* Отменить отложенные запросы канала и немедленно убрать его из кеша (локальное удаление). +/* Отменить JOIN,waiters,предложения и отложенные запросы канала;немедленно убрать его из кеша (локальное удаление). * Активные Merkle-сессии отменяет вызывающий код через member_sync_cancel_channel(). */ void chat_sync_on_channel_deleted(struct UTUN_INSTANCE* inst, const char* ch_id); diff --git a/tests/test_chat_sync_schedule.c b/tests/test_chat_sync_schedule.c index 2f3c656d..ad430f35 100644 --- a/tests/test_chat_sync_schedule.c +++ b/tests/test_chat_sync_schedule.c @@ -4,6 +4,7 @@ #include "transport_layer/etcp_api.h" #include "../lib/u_async.h" #include "../lib/ll_queue.h" +#include "../lib/mem.h" #include "test_clock.h" #include #include @@ -12,13 +13,29 @@ static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg); static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry); static void* schedule_timeout(struct UASYNC* ua, int tb, void* arg, void (*cb)(void*), const char* name); +static void* cache_malloc(uint32_t bytes, const char* site); +static void* cache_realloc(void* old, uint32_t bytes, const char* site); #define member_sync_start observe_start #define etcp_send discard_send #define uasync_set_timeout schedule_timeout +#define u_malloc_impl cache_malloc +#define u_realloc_impl cache_realloc #include "../src/chat/chat_sync.c" #undef member_sync_start #undef etcp_send #undef uasync_set_timeout +#undef u_malloc_impl +#undef u_realloc_impl + +static unsigned cache_allocations, fail_cache_allocation; +static void* cache_malloc(uint32_t bytes, const char* site) { + if (++cache_allocations == fail_cache_allocation) return NULL; + return u_malloc_impl(bytes, site); +} +static void* cache_realloc(void* old, uint32_t bytes, const char* site) { + if (++cache_allocations == fail_cache_allocation) return NULL; + return u_realloc_impl(old, bytes, site); +} struct fixture { struct UASYNC* ua; @@ -30,7 +47,7 @@ struct fixture { static struct { uint64_t peer, channel; } starts[128]; static unsigned start_count; -static int fail_timer; +static int fail_timer, fail_refresh_timer; static struct fixture* injecting; static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg) { @@ -53,6 +70,7 @@ static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry) { static void* schedule_timeout(struct UASYNC* ua, int tb, void* arg, void (*cb)(void*), const char* name) { if (fail_timer && !strcmp(name, "cs_sync")) { fail_timer = 0; return NULL; } + if (fail_refresh_timer && !strcmp(name, "cs_refresh")) { fail_refresh_timer = 0; return NULL; } return uasync_set_timeout(ua, tb, arg, cb, name); } @@ -190,9 +208,49 @@ static void test_lifecycle_and_timer_failure(void) { chat_service_stop(f.inst); run_ms(&f, 1100); assert(!start_count); assert(chat_service_start(f.inst) == 0); assert(!start_count); run_ms(&f, 1100); assert(start_count == 4); + chat_service_stop(f.inst); fail_refresh_timer = 1; + assert(chat_service_start(f.inst) < 0 && !f.inst->chat_sync && !fail_refresh_timer); + assert(!chat_service_start(f.inst)); cleanup(&f); puts("PASS: timer failure recovers; channel/stop cancellation and startup retain correct scope"); } +static struct fixture* deleting; +static void delete_from_event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len) { + if (!deleting || type != CHAT_EVT_MEMBER_UPDATED) return; + assert(len > 1 && data[0] < 64); + char channel[64]; memcpy(channel, data + 1, data[0]); channel[data[0]] = 0; + deleting = NULL; chat_core_delete_channel(inst, channel); +} + +static void test_cache_snapshot(void) { + struct fixture f; setup(&f); + char sql[512]; + snprintf(sql, sizeof(sql), "WITH RECURSIVE n(id) AS (VALUES(1) UNION ALL SELECT id+1 FROM n WHERE id<300)" + " INSERT INTO peers_%s(node_id,x25519_pubkey,ed25519_pubkey,join_sig,join_ts)" + " SELECT id,zeroblob(32),zeroblob(32),zeroblob(64),1 FROM n", f.channel[0]); + assert(sqlite3_exec(f.inst->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK); + cache_allocations = 0; assert(!cs_refresh_channels(f.inst->chat_sync)); + unsigned allocations = cache_allocations; assert(allocations > 4); + struct chat_sync* cs = f.inst->chat_sync; + int largest = 0; for (int i = 0; i < cs->channel_count; i++) if (cs->channels[i].peer_count > largest) largest = cs->channels[i].peer_count; + assert(largest >= 303); + for (unsigned i = 1; i <= allocations; i++) { + struct channel_cache* old = cs->channels; int count = cs->channel_count; + cache_allocations = 0; fail_cache_allocation = i; + assert(cs_refresh_channels(cs) < 0 && cs->channels == old && cs->channel_count == count); + fail_cache_allocation = 0; assert(!cs_refresh_channels(cs)); + } + cs->pending_inv_in_count = 4; + for (int i = 0; i < 4; i++) { cs->pending_inv_in_ch_id[i] = f.gid[i % 2]; cs->pending_inv_in_node_id[i] = 99; } + chat_sync_deny_invite(f.inst, f.gid[0], 99); assert(cs->pending_inv_in_count == 2); + chat_sync_on_channel_deleted(f.inst, f.channel[1]); assert(!cs->pending_inv_in_count); + assert(!cs_refresh_channels(cs)); + deleting = &f; chat_event_set_handler(f.inst, delete_from_event); + cs_on_peer_status_changed(cs, f.peers[1]->peer_node_id, 1); assert(!deleting); + chat_event_set_handler(f.inst, NULL); + cleanup(&f); printf("PASS: 300 peers, %u cache allocation failures/retries, callback deletion and pending invite cleanup\n", allocations); +} + int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG); @@ -200,6 +258,7 @@ int main(void) { size_t baseline = u_get_allocated_count(); test_reinit_isolation(); test_coalesce_and_deadline(); test_sleep_and_cancel(); test_join_existing_transport(); test_reentrant_cancel(); test_lifecycle_and_timer_failure(); + test_cache_snapshot(); test_clock_finish(); assert(u_get_allocated_count() == baseline); puts("ALL PASS: targeted chat synchronization scheduling"); return 0;