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

fix(chat): publish complete peer caches and clean pending JOIN state

Browse Source
This commit is contained in:
evgeny committed 2026-10-27 11:48:09 +01:00
1 parent ce3287ac94
commit 7c050eb7ef
3 files changed
+259 -248
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_sync.c
+191 -236
View File
@@ -35,7 +35,6 @@
#include <string.h>
#include <sqlite3.h>
#include <openssl/evp.h>
#include <openssl/rand.h>
#ifdef __ANDROID__
#include <android/log.h>
@@ -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 */
}
src/chat/chat_sync.h
+8 -11
View File
@@ -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);
tests/test_chat_sync_schedule.c
+60 -1
View File
@@ -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 <assert.h>
#include <stdio.h>
@@ -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;