diff --git a/src/chat/chat_sync.c b/src/chat/chat_sync.c index 6525e2fd..b86b4117 100644 --- a/src/chat/chat_sync.c +++ b/src/chat/chat_sync.c @@ -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); diff --git a/src/chat/merkle_sync.c b/src/chat/merkle_sync.c index 958ce7a1..3b36bae5 100644 --- a/src/chat/merkle_sync.c +++ b/src/chat/merkle_sync.c @@ -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;