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

Scope deferred chat synchronization to channel and peer

Browse Source
This commit is contained in:
evgeny committed 2026-10-25 16:11:11 +01:00
1 parent 5b5221189c
commit 2eade0a7c0
2 files changed
+147 -56
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/chat_sync.c
+132 -54
View File
@@ -49,7 +49,13 @@ struct channel_cache {
int peer_count;
uint32_t msg_count;
uint8_t last_chain_hash[32];
uint8_t synced;
};
/* Только отложенный запрос запуска; состоянием активного обмена владеет Merkle. */
struct pending_sync {
struct pending_sync* next;
uint64_t channel_id, peer_id;
const char* reason; /* статическая строка вызывающего обработчика */
};
struct join_pending {
@@ -68,9 +74,10 @@ struct chat_sync {
void* info_req_timer;
void* join_timer;
void* sync_timer;
struct pending_sync* sync_pending;
uint8_t sync_draining; /* новые задания во время flush обслужит следующий проход */
void* invite_send_timer; /* timer waiting for joiner to respond to CHANNEL_INVITE */
uint8_t initialized;
uint8_t sync_scheduled;
uint64_t pending_invite_ch_id;
uint64_t pending_invite_node_id;
uint8_t pending_invite_is_inviter; /* 1 = inviter sends CHANNEL_INVITE on conn_up, 0 = joiner sends JOIN_INFO_REQ */
@@ -424,39 +431,122 @@ static void cs_post_channel_online(struct chat_sync* cs, const char* ch_id) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: PEERS_ONLINE ch=%.*s total_online=%d", CS_ID, (int)cl, ch_id, online);
}
/* ── Throttled sync (не чаще CS_SYNC_INTERVAL_MS) ── */
/* ── Адресные запросы синхронизации; один таймер на накопленные пары ── */
static void _on_member_sync_done(uint64_t peer, const char* ns, int result, void* arg);
static void cs_sync_timer_cb(void* arg);
static void cs_flush_sync(struct chat_sync* cs) {
uint64_t myid = cs->inst->node_id;
for (int i = 0; i < cs->channel_count; i++) {
struct channel_cache* ch = &cs->channels[i];
for (int j = 0; j < ch->peer_count; j++) {
uint64_t pid = ch->peer_ids[j];
if (pid == myid || !cs_is_peer_online(cs->inst, pid)) continue;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: flush_sync start for peer=%016llx ch=%s",
CS_ID, (unsigned long long)pid, ch->channel_id);
member_sync_start(cs->inst, pid, ch->channel_id, _on_member_sync_done, cs);
}
/* Членство читается из БД, как list_peers: кеш GUI может ещё не знать нового участника. */
static int cs_sync_member(struct chat_sync* cs, uint64_t channel_id, uint64_t peer) {
struct TOPO_GROUP* group = topo_groups_find(cs->inst->topo_groups, channel_id);
if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT) return 0;
char sql[160]; sqlite3_stmt* st = NULL;
snprintf(sql, sizeof(sql), "SELECT 1 FROM peers_%llu WHERE node_id=? AND deleted=0", (unsigned long long)channel_id);
int rc = sqlite3_prepare_v2(cs->inst->topo_sqlite_db, sql, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)peer);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
int found = rc == SQLITE_ROW;
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;
}
static int cs_sync_peer_available(struct chat_sync* cs, uint64_t peer) {
struct ETCP_CONN* conn = instance_find_conn(cs->inst, peer);
return conn && peer != cs->inst->node_id && conn->initialized && conn->links_up && conn->state != 2
&& !conn->close_requested && !conn->reinit_pending && !conn->session_required && !conn->peer_sleep_phase;
}
/* Нулевой peer/channel выбирает все задания соответствующего измерения. Сессии Merkle здесь не трогаем. */
static void cs_cancel_sync(struct chat_sync* cs, uint64_t peer, uint64_t channel_id, const char* reason) {
struct pending_sync** p = &cs->sync_pending;
while (*p) {
struct pending_sync* job = *p;
if ((peer && job->peer_id != peer) || (channel_id && job->channel_id != channel_id)) { p = &job->next; continue; }
*p = job->next;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "sync request cancelled ch=%llu peer=%016llx reason=%s",
(unsigned long long)job->channel_id, (unsigned long long)job->peer_id, reason);
u_free(job);
}
if (!cs->sync_pending && cs->sync_timer) {
uasync_cancel_timeout(cs->inst->ua, cs->sync_timer); cs->sync_timer = NULL;
}
}
static void cs_sync_timer_cb(void* arg) {
struct chat_sync* cs = (struct chat_sync*)arg;
if (!cs || !cs->initialized) return;
cs->sync_timer = NULL;
cs->sync_scheduled = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: sync timer fire", CS_ID);
cs_flush_sync(cs);
/* Срок первого задания не переносится последующими событиями других пиров. */
static int cs_arm_sync(struct chat_sync* cs) {
if (!cs->sync_pending || cs->sync_timer || cs->sync_draining) return 0;
cs->sync_timer = uasync_set_timeout(cs->inst->ua, CS_SYNC_INTERVAL_MS * 10, cs, cs_sync_timer_cb, "cs_sync");
if (cs->sync_timer) return 0;
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "sync request timer allocation failed; cancelling pending requests");
cs_cancel_sync(cs, 0, 0, "timer_failed");
return -1;
}
static void cs_schedule_sync(struct chat_sync* cs) {
if (cs->sync_scheduled) return;
cs->sync_scheduled = 1;
cs->sync_timer = uasync_set_timeout(cs->inst->ua,
CS_SYNC_INTERVAL_MS * 10, cs, cs_sync_timer_cb, "cs_sync");
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: sync scheduled in %ums", CS_ID, CS_SYNC_INTERVAL_MS);
/* Извлекаем задание до callbacks; отмена видит все оставшиеся, новые идут в конец очереди. */
static void cs_sync_timer_cb(void* arg) {
struct chat_sync* cs = arg;
cs->sync_timer = NULL;
if (!cs->initialized) return;
size_t count = 0;
for (struct pending_sync* job = cs->sync_pending; job; job = job->next) count++;
cs->sync_draining = 1;
while (count-- && cs->sync_pending) {
struct pending_sync* job = cs->sync_pending;
cs->sync_pending = job->next;
uint64_t peer = job->peer_id, channel_id = job->channel_id;
const char* reason = job->reason;
u_free(job);
if (!cs_sync_peer_available(cs, peer) || !cs_sync_member(cs, channel_id, peer)) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "sync request discarded ch=%llu peer=%016llx reason=%s unavailable",
(unsigned long long)channel_id, (unsigned long long)peer, reason);
continue;
}
char channel[21]; snprintf(channel, sizeof(channel), "%llu", (unsigned long long)channel_id);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "sync request started ch=%s peer=%016llx reason=%s", channel,
(unsigned long long)peer, reason);
if (member_sync_start(cs->inst, peer, channel, _on_member_sync_done, cs) < 0)
DEBUG_WARN(DEBUG_CATEGORY_MEMBER_SYNC, "sync request rejected ch=%s peer=%016llx reason=%s", channel,
(unsigned long long)peer, reason);
}
cs->sync_draining = 0;
cs_arm_sync(cs);
}
/* Запрос одной пары: объединяем дубликаты, без ссылок на транспорт и заменяемый кеш. */
static int cs_schedule_sync(struct chat_sync* cs, uint64_t channel_id, uint64_t peer, const char* reason) {
if (!cs->initialized || !channel_id || !peer || !cs_sync_peer_available(cs, peer)
|| !cs_sync_member(cs, channel_id, peer)) return 0;
struct pending_sync** tail = &cs->sync_pending;
while (*tail) {
if ((*tail)->channel_id == channel_id && (*tail)->peer_id == peer) {
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "sync request coalesced ch=%llu peer=%016llx reason=%s",
(unsigned long long)channel_id, (unsigned long long)peer, reason);
return 0;
}
tail = &(*tail)->next;
}
struct pending_sync* job = u_calloc(1, sizeof(*job));
if (!job) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "sync request allocation failed ch=%llu peer=%016llx reason=%s",
(unsigned long long)channel_id, (unsigned long long)peer, reason); return -1;
}
job->channel_id = channel_id; job->peer_id = peer; job->reason = reason; *tail = job;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "sync request queued ch=%llu peer=%016llx reason=%s",
(unsigned long long)channel_id, (unsigned long long)peer, reason);
return cs_arm_sync(cs);
}
/* Событие транспорта относится только к общим каналам этого пира; группы актуальнее кеша. */
static void cs_schedule_peer_sync(struct chat_sync* cs, uint64_t peer, const char* reason) {
if (!cs_sync_peer_available(cs, peer)) return;
struct TOPO_GROUPS* groups = cs->inst->topo_groups;
for (struct ll_entry* e = groups && groups->group_list ? groups->group_list->head : NULL; e; e = e->next) {
struct TOPO_GROUP* group = (struct TOPO_GROUP*)e;
if (group->group_type == TOPO_GROUP_TYPE_CHAT) cs_schedule_sync(cs, group->group_id, peer, reason);
}
}
/* ── Peer sleep (keepalive-протокол): спячка → отменить сессии; пробуждение → пересинхронизация ── */
@@ -465,11 +555,12 @@ static void cs_on_peer_sleep(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
struct chat_sync* cs = (struct chat_sync*)arg;
if (!cs || !cs->initialized) return;
if (sleeping) {
cs_cancel_sync(cs, peer_node_id, 0, "peer_sleep");
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: peer sleep peer=0x%016llx — cancel member_sync sessions", CS_ID, (unsigned long long)peer_node_id);
merkle_sync_cancel_peer(inst, peer_node_id);
} else {
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: peer awake peer=0x%016llx — schedule sync", CS_ID, (unsigned long long)peer_node_id);
cs_schedule_sync(cs);
cs_schedule_peer_sync(cs, peer_node_id, "peer_awake");
}
}
@@ -502,7 +593,7 @@ static void cs_send_all_sound_settings(struct chat_sync* cs, uint64_t peer) {
}
}
/* ── Peer status change (GUI + online-бар + schedule_sync; онлайн в БД не хранится) ── */
/* ── Peer status change (только GUI + online-бар; онлайн в БД не хранится) ── */
static void cs_on_peer_status_changed(struct chat_sync* cs, uint64_t peer, int online) {
/* 1. GUI + online-бар */
@@ -528,8 +619,6 @@ static void cs_on_peer_status_changed(struct chat_sync* cs, uint64_t peer, int o
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);
/* 2. Запланировать member_sync_start (throttled) */
cs_schedule_sync(cs);
}
static void cs_resume_db_sync(struct chat_sync* cs, const char* ch_id) {
@@ -548,8 +637,6 @@ static void cs_feedback_done(uint64_t peer, const char* channel, int result, voi
static void _on_member_sync_done(uint64_t peer, const char* ns, int result, void* arg) {
struct chat_sync* cs = (struct chat_sync*)arg;
if (result == MT_OK) {
struct channel_cache* ch = cs ? cs_find(cs, ns) : NULL;
if (ch) ch->synced = CS_SYNC_DONE;
cs_resume_db_sync(cs, ns);
if (cs && merkle_sync_start(cs->inst, MT_DATA_FEEDBACK, peer, ns, cs_feedback_done, cs) < 0)
DEBUG_WARN(DEBUG_CATEGORY_CHAT_SYNC, "%s: feedback sync unavailable ch=%s peer=%016llx", CS_ID, ns, (unsigned long long)peer);
@@ -607,7 +694,7 @@ static void cs_start_channel_join(struct chat_sync* cs, uint64_t peer) {
/* ── Connection callbacks ── */
static void cs_on_conn_up(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
static void cs_on_conn_up(struct ETCP_CONN* conn, int event, void* arg) {
struct chat_sync* cs = (struct chat_sync*)arg;
if (!conn || !cs) return;
uint64_t peer = conn->peer_node_id;
@@ -616,6 +703,7 @@ static void cs_on_conn_up(struct ETCP_CONN* conn, int event, void* arg) { (void)
if (!conn->initialized) { DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: conn_up SKIP — not initialized peer=%016llx", CS_ID, (unsigned long long)peer); return; }
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: conn_up peer=%016llx pending_invite=%llu links_up=%d", CS_ID, (unsigned long long)peer, (unsigned long long)cs->pending_invite_ch_id, conn->links_up);
cs_schedule_peer_sync(cs, peer, event == ETCP_CBK_EVENT_REINIT ? "reinit" : "transport_up");
if (cs->pending_invite_ch_id != 0 && cs->pending_invite_node_id == peer) {
if (cs->join_transport_timer) {
@@ -649,6 +737,7 @@ static void cs_on_conn_down(struct ETCP_CONN* conn, int event, void* arg) { (voi
if (!conn || !cs) return;
uint64_t peer = conn->peer_node_id;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: conn_down peer=%016llx", CS_ID, (unsigned long long)peer);
cs_cancel_sync(cs, peer, 0, "transport_down");
cs_on_peer_status_changed(cs, peer, 0);
@@ -661,13 +750,6 @@ static void cs_on_conn_down(struct ETCP_CONN* conn, int event, void* arg) { (voi
CS_ID, (unsigned long long)peer);
/* invite state survives connection flaps: only timeout or explicit protocol end clears it */
}
for (int i = 0; i < cs->channel_count; i++) {
int found = 0;
for (int j = 0; j < cs->channels[i].peer_count; j++)
if (cs->channels[i].peer_ids[j] == peer) { found = 1; break; }
if (found)
cs->channels[i].synced = CS_SYNC_NONE;
}
}
static void cs_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
@@ -675,7 +757,7 @@ static void cs_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
case ETCP_CONN_STATUS_UP: cs_on_conn_up(conn, ETCP_CBK_EVENT_UP, arg); break;
case ETCP_CONN_STATUS_REINIT:
cs_on_conn_down(conn, ETCP_CBK_EVENT_DOWN, arg);
cs_on_conn_up(conn, ETCP_CBK_EVENT_UP, arg);
cs_on_conn_up(conn, ETCP_CBK_EVENT_REINIT, arg);
break;
case ETCP_CONN_STATUS_DOWN:
case ETCP_CONN_STATUS_DELETE: cs_on_conn_down(conn, ETCP_CBK_EVENT_DOWN, arg); break;
@@ -787,9 +869,9 @@ int chat_sync_init(struct UTUN_INSTANCE* inst) {
chat_join_init(inst);
/* После restart транспорт может уже быть UP: нового conn_up не будет. */
if (cs->channel_count) {
cs_schedule_sync(cs);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: initial member sync scheduled channels=%d", CS_ID, cs->channel_count);
for (struct ll_entry* e = inst->connections ? inst->connections->head : NULL; e; e = e->next) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
cs_schedule_peer_sync(cs, ce->peer_node_id, "startup");
}
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: initialized", CS_ID);
@@ -800,9 +882,8 @@ void chat_sync_destroy(struct UTUN_INSTANCE* inst) {
struct chat_sync* cs = cs_of(inst);
if (!cs || !inst) return;
DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_SYNC, "%s: destroy", CS_ID);
if (cs->sync_timer) { uasync_cancel_timeout(inst->ua, cs->sync_timer); cs->sync_timer = NULL; }
cs_cancel_sync(cs, 0, 0, "service_stop");
if (cs->invite_send_timer) { uasync_cancel_timeout(inst->ua, cs->invite_send_timer); cs->invite_send_timer = NULL; }
if (cs->sync_scheduled) { cs->sync_scheduled = 0; cs_flush_sync(cs); }
member_sync_remove_apply_cbk(inst, cs_on_member_applied, cs);
utun_remove_peer_sleep_cbk(inst, cs_on_peer_sleep, cs);
cs_pending_free_all(cs);
@@ -845,6 +926,7 @@ void chat_sync_connect_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
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->pending_invite_ch_id == strtoull(ch_id, NULL, 10)) {
cs_cancel_proto_timers(cs);
cs->pending_invite_ch_id = 0; cs->pending_invite_node_id = 0;
@@ -1379,6 +1461,7 @@ static void cs_send_join_ready(struct chat_sync* cs, const char* ch_id, uint64_t
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_READY sent to 0x%016llx ch=%s",
CS_ID, (unsigned long long)node_id, ch_id);
cs_pending_remove(cs, node_id, ch_id);
cs_schedule_sync(cs, group_id, node_id, "join_accepted");
}
/* ─── коллбэк member_sync: пришёл рекорд мембера → отправить JOIN_READY джойнеру ─── */
@@ -1457,10 +1540,6 @@ static void cs_handle_join_request(struct chat_sync* cs, uint64_t peer,
cs_refresh_channels(cs);
cs_post_channel_online(cs, ch_id);
/* Новый канал может использовать уже открытый транспорт: события conn_up не будет. */
cs_schedule_sync(cs);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_REQUEST schedule member sync ch=%s peer=%016llx",
CS_ID, ch_id, (unsigned long long)peer);
}
/* ─── JOIN_READY (0x13): connection-узел → joiner (можно запускать BGP) ─── */
@@ -1483,9 +1562,8 @@ static void cs_handle_join_ready(struct chat_sync* cs, uint64_t peer,
}
chat_core_ensure_channel_ready(cs->inst, ch_id);
cs_refresh_channels(cs);
cs_schedule_sync(cs);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "%s: JOIN_READY schedule member sync ch=%s peer=%016llx",
CS_ID, ch_id, (unsigned long long)peer);
/* Транспорт может быть открыт до добавления участника, без нового UP. */
cs_schedule_sync(cs, strtoull(ch_id, NULL, 10), peer, "join_ready");
uint8_t evt[65]; uint8_t ch_id_len = (uint8_t)strlen(ch_id);
evt[0] = ch_id_len; memcpy(evt + 1, ch_id, ch_id_len);
src/chat/merkle_sync.c
+15 -2
View File
@@ -343,8 +343,12 @@ static void ms_work(void* arg) {
s->work = NULL;
if (s->state != MS_IDLE || !s->dirty || s->tx) return;
int rc;
if (s->leader) rc = ms_begin(s);
else { s->dirty = 0; rc = ms_send(s, MS_WAKE, 0, NULL, 0); }
if (s->leader) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "merkle local work: dataset=%u ns=%s peer=%016llx revision=%llu verified=%llu",
s->dataset->id, s->ns, (unsigned long long)s->peer,
(unsigned long long)s->revision, (unsigned long long)s->verified_revision);
rc = ms_begin(s);
} else { s->dirty = 0; rc = ms_send(s, MS_WAKE, 0, NULL, 0); }
if (rc < 0) ms_fail(s, MT_ERR_IO, "start send", 1);
}
@@ -354,6 +358,8 @@ void merkle_sync_changed(struct UTUN_INSTANCE* inst, uint8_t dataset, const char
for (struct ms_session* s = ms->sessions; s; s = s->next) {
if (s->dataset->id != dataset || strcmp(s->ns, ns)) continue;
s->revision++; s->dirty = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "merkle changed: dataset=%u ns=%s peer=%016llx state=%d revision=%llu",
dataset, ns, (unsigned long long)s->peer, s->state, (unsigned long long)s->revision);
if (s->state == MS_FAILED) { s->state = MS_IDLE; s->previous_valid = 0; }
if (s->state == MS_IDLE && ms_schedule(s) < 0) {
ms_fail(s, MT_ERR_IO, "change scheduling", 1); return;
@@ -500,6 +506,10 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
/* Follower мог отменить и заново подключить namespace. Старый запрос уже некому обслужить. */
ms_discard_tx(s);
s->previous_valid = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
"merkle remote wake: dataset=%u ns=%s peer=%016llx state=%d dirty=%d revision=%llu verified=%llu",
dataset, ns, (unsigned long long)s->peer, s->state, s->dirty,
(unsigned long long)s->revision, (unsigned long long)s->verified_revision);
rc = ms_begin(s);
}
} else if (type == MS_BEGIN) {
@@ -657,6 +667,9 @@ int merkle_sync_start(struct UTUN_INSTANCE* inst, uint8_t dataset, uint64_t peer
u_free(r); return -1;
}
if (s->state == MS_FAILED) { s->state = MS_IDLE; s->previous_valid = 0; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
"merkle start request: dataset=%u ns=%s peer=%016llx state=%d dirty=%d revision=%llu callback=%d",
dataset, ns, (unsigned long long)peer, s->state, s->dirty, (unsigned long long)s->revision, cb != NULL);
s->dirty = 1;
if (s->state == MS_IDLE && ms_schedule(s) < 0) {
u_free(r); ms_fail(s, MT_ERR_IO, "start scheduling", 1); return -1;