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

chat: локальное удаление канала (Android) + db_sync инстансы в куче

Browse Source
This commit is contained in:
evgeny committed 2026-09-13 22:08:33 +03:00
1 parent ddfffad1ed
commit 7c8d65db78
30 files changed
+466 -42
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/db_sync.c
+32 -18
View File
@@ -82,7 +82,7 @@ struct DB_SYNC {
sqlite3* db;
uint8_t shared_db; /* db == inst->topo_sqlite_db, do not close */
uint64_t last_connected_tb;
struct DB_SYNC_INSTANCE* instances;
struct DB_SYNC_INSTANCE** instances;
int instance_count, instance_capacity;
void* peer_check_timer;
void* peer_check_wait; /* standby_wait handle (Android) */
@@ -225,26 +225,33 @@ static void db_chain_hash_compute(const uint8_t prev_chain_hash[32],
static struct DB_SYNC_INSTANCE* db_instance_find(struct DB_SYNC* db, uint64_t group_id)
{
for (int i = 0; i < db->instance_count; i++) {
if (db->instances[i].group_id == group_id && db->instances[i].enabled) return &db->instances[i];
if (db->instances[i]->group_id == group_id && db->instances[i]->enabled) return db->instances[i];
}
return NULL;
}
// Выделяет новый слот инстанса в массиве, расширяя его при необходимости
struct DB_SYNC_INSTANCE* db_sync_instance_find(struct UTUN_INSTANCE* inst, uint64_t group_id)
{
if (!inst || !inst->db_sync) return NULL;
return db_instance_find(inst->db_sync, group_id);
}
// Выделяет новый инстанс в куче (стабильный указатель) и регистрирует его в массиве
static struct DB_SYNC_INSTANCE* db_instance_alloc(struct DB_SYNC* db)
{
DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "count=%d cap=%d", db->instance_count, db->instance_capacity);
if (db->instance_count >= db->instance_capacity) {
int nc = db->instance_capacity ? db->instance_capacity * 2 : 4;
struct DB_SYNC_INSTANCE* np = u_realloc(db->instances, nc * sizeof(*db->instances));
struct DB_SYNC_INSTANCE** np = u_realloc(db->instances, nc * sizeof(*db->instances));
if (!np) return NULL;
db->instances = np;
db->instance_capacity = nc;
}
struct DB_SYNC_INSTANCE* si = &db->instances[db->instance_count++];
memset(si, 0, sizeof(*si));
struct DB_SYNC_INSTANCE* si = u_calloc(1, sizeof(*si));
if (!si) return NULL;
si->db_sync = db;
si->enabled = 1;
db->instances[db->instance_count++] = si;
return si;
}
@@ -856,8 +863,10 @@ static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si, uint64_t src, const
struct SI_PEER* sp = si_peer_find(si, src);
// Peer empty → send ALL our data, then SYNC_DONE will be sent from DATA handler when slave responds
if (peer_ch8 == 0 && sc == 0) {
// Peer empty → send ALL our data, then SYNC_DONE will be sent from DATA handler when slave responds.
// Если мы тоже пусты (mc==0) — обе стороны пусты: не отправляем ничего, а завершаем sync через
// SYNC_DONE в ветке "exact match" ниже (иначе обе стороны вечно ждут ответа → TIMEOUT).
if (peer_ch8 == 0 && sc == 0 && mc > 0) {
uint32_t vp = (uint32_t)-1; if (sp) sp->verified_pos = vp;
uint32_t sent = 0;
while (sent < mc) {
@@ -1293,7 +1302,7 @@ static void db_sync_on_conn_up(struct ETCP_CONN* conn, void* arg)
int synced = 0;
char tbl_list[256] = "";
for (int i = 0; i < db->instance_count; i++) {
struct DB_SYNC_INSTANCE* si = &db->instances[i];
struct DB_SYNC_INSTANCE* si = db->instances[i];
if (!si->enabled) continue;
struct SI_PEER* p = si_peer_add(si, pid);
if (!p) continue;
@@ -1323,14 +1332,14 @@ static void db_sync_on_conn_down(struct ETCP_CONN* conn, void* arg)
DEBUG_TRACE(DEBUG_CATEGORY_CHAT_SYNC, "peer=%016llx", (unsigned long long)pid);
for (int i = 0; i < db->instance_count; i++) {
struct SI_PEER* p = si_peer_find(&db->instances[i], pid);
struct SI_PEER* p = si_peer_find(db->instances[i], pid);
if (p) p->sync_state = 0;
}
int any = 0;
for (int i = 0; i < db->instance_count; i++) {
for (int j = 0; j < db->instances[i].peer_count; j++) {
if (db->instances[i].peers[j].sync_state >= 1) { any = 1; goto cd_done; }
for (int j = 0; j < db->instances[i]->peer_count; j++) {
if (db->instances[i]->peers[j].sync_state >= 1) { any = 1; goto cd_done; }
}
}
cd_done:
@@ -1507,7 +1516,7 @@ static int db_sync_has_pending_work(struct DB_SYNC* db)
{
if (!db || !db->enabled) return 0;
for (int i = 0; i < db->instance_count; i++) {
struct DB_SYNC_INSTANCE* si = &db->instances[i];
struct DB_SYNC_INSTANCE* si = db->instances[i];
if (!si->enabled) continue;
for (int j = 0; j < si->peer_count; j++)
if (si->peers[j].sync_state != 2) return 1;
@@ -1546,7 +1555,7 @@ static void db_sync_peer_check_cb(void* arg)
char launched_list[256] = "";
if (g) {
for (int i = 0; i < db->instance_count; i++) {
struct DB_SYNC_INSTANCE* si = &db->instances[i];
struct DB_SYNC_INSTANCE* si = db->instances[i];
if (!si->enabled) continue;
if (si->sync_gated) continue;
@@ -1591,7 +1600,7 @@ static void db_sync_peer_check_cb(void* arg)
uint64_t now = get_time_tb();
uint64_t to_tb = DB_SYNC_SYNC_TIMEOUT * 10000u;
for (int i = 0; i < db->instance_count; i++) {
struct DB_SYNC_INSTANCE* si = &db->instances[i];
struct DB_SYNC_INSTANCE* si = db->instances[i];
if (!si->enabled) continue;
for (int j = 0; j < si->peer_count; j++) {
struct SI_PEER* p = &si->peers[j];
@@ -1722,10 +1731,11 @@ void db_sync_destroy(struct UTUN_INSTANCE* inst)
#endif
for (int i = 0; i < db->instance_count; i++) {
struct DB_SYNC_INSTANCE* si = &db->instances[i];
struct DB_SYNC_INSTANCE* si = db->instances[i];
if (si->ttl_timer) { uasync_cancel_timeout(inst->ua, si->ttl_timer); si->ttl_timer = NULL; }
if (si->recalc.timer) { uasync_cancel_timeout(inst->ua, si->recalc.timer); si->recalc.timer = NULL; }
if (si->peers) u_free(si->peers);
u_free(si);
}
etcp_remove_conn_status_cbk(inst, db_sync_on_conn_status, NULL);
@@ -1834,13 +1844,17 @@ void db_sync_instance_remove(struct DB_SYNC_INSTANCE* si)
if (si->recalc.timer) { uasync_cancel_timeout(inst->ua, si->recalc.timer); si->recalc.timer = NULL; }
if (si->peers) { u_free(si->peers); si->peers = NULL; si->peer_count = si->peer_capacity = 0; }
int idx = (int)(si - db->instances);
if (idx >= 0 && idx < db->instance_count) {
int idx = -1;
for (int i = 0; i < db->instance_count; i++) {
if (db->instances[i] == si) { idx = i; break; }
}
if (idx >= 0) {
memmove(&db->instances[idx], &db->instances[idx + 1],
(db->instance_count - idx - 1) * sizeof(*db->instances));
db->instance_count--;
}
DEBUG_INFO(DEBUG_CATEGORY_CHAT_SYNC, "instance removed tbl=%s group=%016llx", si->table_name, (unsigned long long)si->group_id);
u_free(si);
}
// Вставляет подписанную запись: проверяет подпись, вставляет в БД, рассылает PUSH всем синхронизированным пирам