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

debug: add DEBUG_CATEGORY_DB_SYNC (26) + TRACE logs for db_sync, merkle_sync, chat_sync, member_sync

topo_upd
Evgeny 3 months ago
parent
9e34e9f838
commit
4b7c7738de
7 changed files with 166 additions and 96 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 6
      AGENTS.md
  2. 1
      lib/debug_config.c
  3. 3
      lib/debug_config.h
  4. 59
      src/db_sync.c
  5. 135
      tools/chatgui/transport/chat_sync.c
  6. 15
      tools/chatgui/transport/member_sync.c
  7. 43
      tools/chatgui/transport/merkle_sync.c

6
AGENTS.md
Unescape View File

@ -196,8 +196,12 @@ sc_obfuscate_pubkey(salt, peer_pubkey_bin, my_pubkey_bin, obfuscated_output);
## Debug System
### Debug Levels (по возрастанию)
### Debug Levels
`none < error < warn < info < debug < trace`
trace - логируем заходы в функции и важные ветвления внутри функций
debug - всё что нужно для понимание сути происходящего (состояний, важных переменных) алгоритма. тоесть по уровню debug мы должны видеть все важные внутреннии нюансы работы функции включая состояния и переменные.
info - сообщения для пользователя о работе. мы должны видеть в читаемом виде понятные и осмысленные сообщения о каких-либо значимых действиях с точки зрения логики работы модуля или приложения
warn / error - ошибки и предупреждения (аномалии). должны быть во всех ошибочных ветках. Молча вываливаться с ошибкой нельзя, все ошибки и предупреждения должны логироваться.
### Debug Categories (25 категорий)
```

1
lib/debug_config.c
Unescape View File

@ -130,6 +130,7 @@ static const struct {
{"bbr", DEBUG_CATEGORY_BBR},
{"etcp_dump", DEBUG_CATEGORY_ETCP_DUMP},
{"connectivity", DEBUG_CATEGORY_CONNECTIVITY},
{"db_sync", DEBUG_CATEGORY_DB_SYNC},
{"all", DEBUG_CATEGORY_ALL},
{NULL, DEBUG_CATEGORY_NONE}
};

3
lib/debug_config.h
Unescape View File

@ -64,7 +64,8 @@ typedef int debug_category_t;
#define DEBUG_CATEGORY_BBR 23 // BBR congestion control
#define DEBUG_CATEGORY_ETCP_DUMP 24 // ETCP packet dump
#define DEBUG_CATEGORY_CONNECTIVITY 25 // Connection/handshake lifecycle
#define DEBUG_CATEGORY_COUNT 26 // Total number of categories
#define DEBUG_CATEGORY_DB_SYNC 26 // DB sync (chat, members, distributed tables)
#define DEBUG_CATEGORY_COUNT 27 // Total number of categories
#define DEBUG_CATEGORY_ALL (-1) // special value for all categories
/* Debug configuration structure */

59
src/db_sync.c
Unescape View File

@ -16,8 +16,6 @@
#include <string.h>
#include <openssl/evp.h>
#define DEBUG_CATEGORY_DB_SYNC DEBUG_CATEGORY_DEBUG
// ---- Forward declarations ----
struct DB_SYNC;
struct DB_SYNC_INSTANCE;
@ -91,6 +89,7 @@ static int si_prep(struct DB_SYNC_INSTANCE* si, sqlite3_stmt** stmt, const char*
static int db_sqlite_open(struct DB_SYNC* db, const char* path)
{
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "path=%s", path);
int rc = sqlite3_open(path, &db->db);
if (rc != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "db_sync: sqlite3_open(%s): %s", path, sqlite3_errmsg(db->db));
@ -109,6 +108,7 @@ static int db_sqlite_open(struct DB_SYNC* db, const char* path)
static void db_sqlite_close(struct DB_SYNC* db)
{
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "");
if (!db->db) return;
sqlite3_close(db->db);
db->db = NULL;
@ -162,6 +162,7 @@ static struct DB_SYNC_INSTANCE* db_instance_find(struct DB_SYNC* db, uint64_t ha
static struct DB_SYNC_INSTANCE* db_instance_alloc(struct DB_SYNC* db)
{
DEBUG_TRACE(DEBUG_CATEGORY_DB_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));
@ -297,6 +298,7 @@ static int db_prev_chain_hash(struct DB_SYNC_INSTANCE* si, uint64_t ts, const ui
static void db_cascade_from(struct DB_SYNC_INSTANCE* si, uint32_t from_pos)
{
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "tbl=%s from=%u", SI_TBL(si), from_pos);
sqlite3* db = SI_DB(si);
int rc = sqlite3_exec(db, "BEGIN IMMEDIATE", NULL, NULL, NULL);
if (rc != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "cascade_from BEGIN: %s", sqlite3_errmsg(db)); return; }
@ -412,6 +414,7 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
const char* json, size_t jlen,
const uint8_t* author_sig, int do_cascade)
{
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "id=%llu ts=%llu author=%N len=%zu cascade=%d", (unsigned long long)id, (unsigned long long)ts, (unsigned long long)author_node_id, jlen, do_cascade);
sqlite3* db = SI_DB(si);
sqlite3_stmt* stmt;
int rc;
@ -429,7 +432,7 @@ static int db_record_insert(struct DB_SYNC_INSTANCE* si,
sqlite3_bind_blob(stmt, 2, author_sig, DB_SIG_SIZE, SQLITE_STATIC);
int exists = (sqlite3_step(stmt) == SQLITE_ROW);
sqlite3_finalize(stmt);
if (exists) { sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return 1; }
if (exists) { DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ already exists, skip"); sqlite3_exec(db, "ROLLBACK", NULL, NULL, NULL); return 1; }
// Compute chain hash
uint8_t prev_ch[32];
@ -581,6 +584,7 @@ static void si_delivery_update(struct DB_SYNC_INSTANCE* si, uint64_t ts, uint64_
static void db_handle_error(struct DB_SYNC* db, uint64_t src, const uint8_t* p, size_t len)
{
if (len < 1) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx code=%u", (unsigned long long)src, p[0]);
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "db_sync: ERROR from %016llx code=%u", (unsigned long long)src, p[0]);
(void)db;
}
@ -590,8 +594,7 @@ static void db_handle_init_sync(struct DB_SYNC_INSTANCE* si, uint64_t src, const
if (len < 4) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "INIT_SYNC too short %zu from %016llx", len, (unsigned long long)src); return; }
uint32_t pc = *(uint32_t*)p;
uint32_t mc = db_count(si);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "INIT_SYNC from %016llx peer_count=%u my_count=%u tbl=%s",
(unsigned long long)src, pc, mc, SI_TBL(si));
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx peer_cnt=%u my=%u tbl=%s", (unsigned long long)src, pc, mc, SI_TBL(si));
uint32_t tp = (pc < mc ? pc : mc);
if (tp > 0) tp--;
@ -629,12 +632,14 @@ static void db_handle_init_resp(struct DB_SYNC_INSTANCE* si, uint64_t src, const
uint32_t tp = *(uint32_t*)p;
uint64_t peer_ch8; memcpy(&peer_ch8, p + 4, 8);
uint8_t sc = p[12];
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx tp=%u peer_ch8=%016llx sc=%d", (unsigned long long)src, tp, (unsigned long long)peer_ch8, sc);
uint64_t my_ch8 = 0;
db_chain_hash8_at(si, tp, &my_ch8);
if (peer_ch8 == 0 && sc == 0) {
uint32_t mc = db_count(si);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ peer empty, sending all %u records", mc);
uint32_t batches = (mc + DB_SEND_DATA_MAX - 1) / DB_SEND_DATA_MAX;
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "peer %016llx empty, sending all %u records in %u batches",
(unsigned long long)src, mc, batches);
@ -835,8 +840,10 @@ static void db_handle_refine(struct DB_SYNC_INSTANCE* si, uint64_t src, const ui
uint32_t from = *(uint32_t*)p;
uint32_t to = *(uint32_t*)(p + 4);
uint8_t hc = p[8];
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx range=[%u..%u] hc=%d", (unsigned long long)src, from, to, hc);
if (hc == 0) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ hc=0, sending data batch from %u", from);
uint32_t mc = db_count(si);
uint32_t scnt = (to - from + 1) < DB_SEND_DATA_MAX ? (to - from + 1) : DB_SEND_DATA_MAX;
if (from >= mc) return;
@ -898,6 +905,7 @@ static void db_handle_send_data(struct DB_SYNC_INSTANCE* si, uint64_t src, const
if (len < 6) return;
uint32_t from = *(uint32_t*)p;
uint16_t count = *(uint16_t*)(p + 4);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx from=%u count=%u", (unsigned long long)src, from, count);
const uint8_t* ptr = p + 6;
struct SI_PEER* sp = si_peer_find(si, src);
@ -955,6 +963,9 @@ static void db_handle_sync_done(struct DB_SYNC_INSTANCE* si, uint64_t src, const
uint32_t mc = db_count(si);
uint64_t mch8 = 0;
if (mc > 0) db_chain_hash8_at(si, mc - 1, &mch8);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx my_cnt=%u peer_cnt=%u my_ch8=%016llx peer_ch8=%016llx → %s",
(unsigned long long)src, mc, pc, (unsigned long long)mch8, (unsigned long long)pch8,
(mc == pc && mch8 == pch8) ? "matched" : "MISMATCH");
if (mc != pc || mch8 != pch8) {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
@ -974,6 +985,7 @@ static void db_handle_ack_push(struct DB_SYNC_INSTANCE* si, uint64_t src, const
if (len < 16) return;
uint64_t ts = *(uint64_t*)p;
uint64_t author = *(uint64_t*)(p + 8);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx ts=%llu author=%016llx", (unsigned long long)src, (unsigned long long)ts, (unsigned long long)author);
sqlite3_stmt* stmt;
if (si_prep(si, &stmt,
@ -1003,6 +1015,7 @@ static void db_handle_push(struct DB_SYNC_INSTANCE* si, uint64_t src, const uint
const uint8_t* rdata, *rsig;
int rsiglen;
if (si_parse_record(&ptr, p + len, &rid, &rts, &rauthor, &rdlen, &rdata, &rsig, &rsiglen) != 0) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "from=%016llx id=%llu author=%016llx ts=%llu len=%u", (unsigned long long)src, (unsigned long long)rid, (unsigned long long)rauthor, (unsigned long long)rts, rdlen);
int ret = db_record_insert(si, rid, rts, rauthor, (const char*)rdata, rdlen, rsig, 0);
if (ret == 0) {
@ -1036,12 +1049,14 @@ static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
uint8_t type = entry->dgram[9];
const uint8_t* payload = entry->dgram + 10;
size_t plen = entry->len - 10;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "recv type=%02x from=%016llx hash=%016llx len=%zu", type, (unsigned long long)src, (unsigned long long)hash, plen);
struct DB_SYNC_INSTANCE* si = db_instance_find(db, hash);
if (!si) {
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC,
"recv msg type=0x%02x from %016llx hash=%016llx — instance NOT FOUND; sending DB_ERR_NOT_FOUND",
type, (unsigned long long)src, (unsigned long long)hash);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ instance not found for hash=%016llx", (unsigned long long)hash);
uint8_t err[2]; err[0] = DB_MSG_ERROR; err[1] = DB_ERR_NOT_FOUND;
db_sync_send_hash(db, src, hash, err, 2);
queue_dgram_free(entry); queue_entry_free(entry); return;
@ -1053,14 +1068,14 @@ static void db_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
}
switch (type) {
case DB_MSG_INIT_SYNC: db_handle_init_sync(si, src, payload, plen); break;
case DB_MSG_INIT_RESP: db_handle_init_resp(si, src, payload, plen); break;
case DB_MSG_REFINE: db_handle_refine(si, src, payload, plen); break;
case DB_MSG_SEND_DATA: db_handle_send_data(si, src, payload, plen); break;
case DB_MSG_PUSH: db_handle_push(si, src, payload, plen); break;
case DB_MSG_ACK_PUSH: db_handle_ack_push(si, src, payload, plen); break;
case DB_MSG_SYNC_DONE: db_handle_sync_done(si, src, payload, plen); break;
case DB_MSG_ERROR: db_handle_error(db, src, payload, plen); break;
case DB_MSG_INIT_SYNC: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ handle INIT_SYNC"); db_handle_init_sync(si, src, payload, plen); break;
case DB_MSG_INIT_RESP: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ handle INIT_RESP"); db_handle_init_resp(si, src, payload, plen); break;
case DB_MSG_REFINE: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ handle REFINE"); db_handle_refine(si, src, payload, plen); break;
case DB_MSG_SEND_DATA: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ handle SEND_DATA"); db_handle_send_data(si, src, payload, plen); break;
case DB_MSG_PUSH: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ handle PUSH"); db_handle_push(si, src, payload, plen); break;
case DB_MSG_ACK_PUSH: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ handle ACK_PUSH"); db_handle_ack_push(si, src, payload, plen); break;
case DB_MSG_SYNC_DONE: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ handle SYNC_DONE"); db_handle_sync_done(si, src, payload, plen); break;
case DB_MSG_ERROR: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ handle ERROR"); db_handle_error(db, src, payload, plen); break;
default:
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "unknown msg type 0x%02x from %016llx", type, (unsigned long long)src);
break;
@ -1077,6 +1092,7 @@ static void db_sync_on_new_conn(struct ETCP_CONN* conn, void* arg)
{
(void)arg;
if (!conn || !conn->instance || !conn->instance->db_sync) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "conn=%p node=%016llx", (void*)conn, (unsigned long long)conn->peer_node_id);
etcp_conn_add_up_cbk(conn, db_sync_on_conn_up, NULL);
etcp_conn_add_down_cbk(conn, db_sync_on_conn_down, NULL);
}
@ -1089,6 +1105,7 @@ static void db_sync_on_conn_up(struct ETCP_CONN* conn, void* arg)
if (!db->enabled) return;
uint64_t pid = conn->peer_node_id;
if (pid == 0 || pid == db->inst->node_id) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "peer=%016llx init=%d links=%d", (unsigned long long)pid, conn->initialized, conn->links_up);
db->last_connected_tb = get_time_tb();
int synced = 0, skipped_state = 0, skipped_not_ready = 0;
@ -1117,6 +1134,7 @@ static void db_sync_on_conn_down(struct ETCP_CONN* conn, void* arg)
if (!db->enabled) return;
uint64_t pid = conn->peer_node_id;
if (pid == 0) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_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);
@ -1141,7 +1159,7 @@ cd_done:
static void db_sync_initiate_sync(struct DB_SYNC_INSTANCE* si, uint64_t pid)
{
uint32_t mc = db_count(si);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "INIT_SYNC → %016llx my=%u tbl=%s", (unsigned long long)pid, mc, SI_TBL(si));
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "pid=%016llx my=%u tbl=%s", (unsigned long long)pid, mc, SI_TBL(si));
uint8_t msg[5];
msg[0] = DB_MSG_INIT_SYNC;
@ -1153,6 +1171,7 @@ static void db_sync_initiate_sync(struct DB_SYNC_INSTANCE* si, uint64_t pid)
static void db_verify_chain(struct DB_SYNC_INSTANCE* si)
{
uint32_t mc = db_count(si);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "tbl=%s mc=%u", SI_TBL(si), mc);
if (mc == 0) return;
uint8_t prev_ch[32]; memset(prev_ch, 0, 32);
@ -1196,6 +1215,7 @@ static void db_sync_peer_check_cb(void* arg)
{
struct DB_SYNC* db = (struct DB_SYNC*)arg;
if (!db || !db->enabled) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "instances=%d", db->instance_count);
struct TOPO_GROUP* g = topo_groups_get_default(db->inst->topo_groups);
if (!g) {
@ -1242,6 +1262,7 @@ static void db_sync_peer_check_cb(void* arg)
static void db_sync_instance_ttl_cb(void* arg)
{
struct DB_SYNC_INSTANCE* si = (struct DB_SYNC_INSTANCE*)arg;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "tbl=%s enabled=%d", SI_TBL(si), si->enabled);
uint64_t interval = DB_SYNC_TTL_INTERVAL * 10000u;
if (!si || !si->enabled || !si->db_sync || !si->db_sync->db) {
@ -1281,6 +1302,7 @@ static void db_sync_instance_ttl_cb(void* arg)
int db_sync_init(struct UTUN_INSTANCE* inst)
{
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "NULL instance"); return -1; }
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "inst=%p enabled=%d", (void*)inst, inst->config->global.db_sync_enabled);
struct DB_SYNC* db = u_calloc(1, sizeof(struct DB_SYNC));
if (!db) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "u_calloc failed"); return -1; }
db->inst = inst;
@ -1331,6 +1353,7 @@ int db_sync_init(struct UTUN_INSTANCE* inst)
void db_sync_destroy(struct UTUN_INSTANCE* inst)
{
if (!inst || !inst->db_sync) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "inst=%p instances=%d", (void*)inst, inst->db_sync->instance_count);
struct DB_SYNC* db = inst->db_sync;
inst->db_sync = NULL;
@ -1365,6 +1388,7 @@ void db_sync_destroy(struct UTUN_INSTANCE* inst)
struct DB_SYNC_INSTANCE* db_sync_instance_add(struct UTUN_INSTANCE* inst, const char* name, uint64_t id)
{
if (!inst || !inst->db_sync || !name || !name[0]) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "instance_add invalid args"); return NULL; }
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "name=%s id=%016llx", name, (unsigned long long)id);
struct DB_SYNC* db = inst->db_sync;
if (!db->enabled || !db->db) return NULL;
@ -1445,6 +1469,7 @@ struct DB_SYNC_INSTANCE* db_sync_instance_add(struct UTUN_INSTANCE* inst, const
void db_sync_instance_remove(struct DB_SYNC_INSTANCE* si)
{
if (!si) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "tbl=%s hash=%016llx", si->table_name, (unsigned long long)si->hash);
struct DB_SYNC* db = si->db_sync;
struct UTUN_INSTANCE* inst = db->inst;
@ -1466,6 +1491,7 @@ int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* json_data, si
{
if (!si || !si->enabled || !json_data || len == 0) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "insert invalid args"); return -1; }
if (!sig || sig_len != DB_SIG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "insert_signed: signature required (64 bytes Ed25519)"); return -1; }
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "tbl=%s id=%llu ts=%llu len=%zu", SI_TBL(si), (unsigned long long)si->next_id, (unsigned long long)ts, len);
uint64_t id = si->next_id;
uint64_t author_node_id = si->db_sync->inst->node_id;
@ -1488,12 +1514,14 @@ int db_sync_insert_signed(struct DB_SYNC_INSTANCE* si, const char* json_data, si
if (off + sl <= sizeof(pbuf)) { memcpy(pbuf + off, sig, sl); off += sl; }
// Push to all synced peers
int push_count = 0;
for (int i = 0; i < si->peer_count; i++) {
if (si->peers[i].sync_state >= 1 && si->peers[i].node_id != si->db_sync->inst->node_id) {
if (db_sync_send(si, si->peers[i].node_id, pbuf, off) >= 0)
si_delivery_update(si, ts, author_node_id, si->peers[i].node_id);
{ si_delivery_update(si, ts, author_node_id, si->peers[i].node_id); push_count++; }
}
}
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "→ pushed to %d synced peers", push_count);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "insert id=%llu author=%016llx ts=%llu len=%zu",
(unsigned long long)id, (unsigned long long)author_node_id, (unsigned long long)ts, len);
@ -1533,6 +1561,7 @@ int db_sync_select(struct DB_SYNC_INSTANCE* si, uint32_t offset, uint32_t limit,
db_sync_select_cb cb, void* arg)
{
if (!si || !si->enabled || !cb) return 0;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "tbl=%s off=%u lim=%u", SI_TBL(si), offset, limit);
sqlite3_stmt* stmt;
if (si_prep(si, &stmt,

135
tools/chatgui/transport/chat_sync.c
Unescape View File

@ -108,7 +108,7 @@ static void ac_gc(struct auto_connect* ac) {
continue;
}
/* expired and not up — cancel */
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTIVITY, "%s: GC closing flight node=0x%016llx (expired)", AC_ID, (unsigned long long)nid);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: GC closing flight node=0x%016llx (expired)", AC_ID, (unsigned long long)nid);
chat_core_connect_auto_cancel(ac->flights[i].ca_state);
memset(&ac->flights[i], 0, sizeof(ac->flights[i]));
}
@ -183,7 +183,7 @@ static void ac_fill(struct auto_connect* ac) {
f->node_id = 0; f->created_tb = 0;
continue;
}
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTIVITY, "%s: flight[%d] launched node=0x%016llx", AC_ID, slot, (unsigned long long)nid);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: flight[%d] launched node=0x%016llx", AC_ID, slot, (unsigned long long)nid);
uint8_t evt[7]; evt[0] = 0; uint16_t t = 0, n = (uint16_t)ac->channel_count, s = 0;
memcpy(evt + 1, &t, 2); memcpy(evt + 3, &n, 2); memcpy(evt + 5, &s, 2);
gui_bridge_post(GUI_EVT_AUTO_CONNECT_STATUS, evt, 7);
@ -196,7 +196,7 @@ static void ac_result_cb(int result, uint64_t node_id, void* arg) {
struct ac_flight* f = (struct ac_flight*)arg;
if (!g_ac || !g_ac->active) return;
const char* rs = result == CC_OK ? "OK" : "FAIL";
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: result node=0x%016llx %s", AC_ID, (unsigned long long)node_id, rs);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: result node=0x%016llx %s", AC_ID, (unsigned long long)node_id, rs);
if (result == CC_OK) f->ca_state = NULL; /* already freed by ca_ready_cb, prevent double-free in GC/stop */
}
@ -225,7 +225,7 @@ void chat_sync_auto_connect_start(struct UTUN_INSTANCE* inst) {
ac->active = 1;
g_ac = ac;
ac_load_channels(ac);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: start, %d channels loaded", AC_ID, ac->channel_count);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: start, %d channels loaded", AC_ID, ac->channel_count);
ac_gc(ac);
ac_fill(ac);
ac->retry_timer = uasync_set_timeout(inst->ua, AC_RETRY_MS * 10, ac, ac_retry_timer_cb, "ac_retry");
@ -243,11 +243,11 @@ void chat_sync_auto_connect_stop(void) {
memcpy(evt + 1, &z, 2); memcpy(evt + 3, &z, 2); memcpy(evt + 5, &z, 2);
gui_bridge_post(GUI_EVT_AUTO_CONNECT_STATUS, evt, 7);
u_free(ac);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: stopped", AC_ID);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: stopped", AC_ID);
}
void chat_sync_auto_connect_switch_group(struct UTUN_INSTANCE* inst, uint64_t new_group_id) {
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: switch group 0x%016llx", AC_ID, (unsigned long long)new_group_id);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: switch group 0x%016llx", AC_ID, (unsigned long long)new_group_id);
chat_sync_auto_connect_stop();
chat_sync_auto_connect_start(inst);
}
@ -322,10 +322,10 @@ static int cs_send(struct chat_sync* cs, const char* ch_id, uint64_t dst,
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { u_free(buf); return -1; }
entry->dgram = buf; entry->len = total;
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: SEND %s to=%016llx ch=%s len=%zu",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: SEND %s to=%016llx ch=%s len=%zu",
CS_ID, cs_msg_name(payload[0]), (unsigned long long)dst, ch_id, len);
struct ETCP_CONN* conn = cs_find_conn_for_node(cs->inst, dst);
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: no conn for node %016llx", CS_ID, (unsigned long long)dst); u_free(buf); queue_entry_free(entry); return -1; }
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: no conn for node %016llx", CS_ID, (unsigned long long)dst); u_free(buf); queue_entry_free(entry); return -1; }
return etcp_send(conn, entry);
}
@ -373,20 +373,22 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
char ch_id[64]; memcpy(ch_id, d + 2, ch_len); ch_id[ch_len] = '\0';
uint8_t type = d[2 + ch_len];
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: recv %s(%02x) from=%016llx ch=%s len=%zu",
CS_ID, cs_msg_name(type), type, (unsigned long long)peer, ch_id, dlen);
const uint8_t* pl = d + 3 + ch_len;
size_t plen = dlen - 3 - ch_len;
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: RECV %s from=%016llx ch=%s len=%zu",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: RECV %s from=%016llx ch=%s len=%zu",
CS_ID, cs_msg_name(type), (unsigned long long)peer, ch_id, plen);
switch (type) {
case CS_MSG_CHANNEL_INFO_REQ: cs_handle_channel_info_req(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_CHANNEL_INFO_RESP:cs_handle_channel_info_resp(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_CHANNEL_JOIN: cs_handle_channel_join(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_WELCOME: cs_handle_welcome(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_PEER_UPSERT: cs_handle_peer_upsert(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_CHANNEL_INFO_REQ: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → handle INFO_REQ", CS_ID); cs_handle_channel_info_req(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_CHANNEL_INFO_RESP:DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → handle INFO_RESP", CS_ID); cs_handle_channel_info_resp(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_CHANNEL_JOIN: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → handle JOIN", CS_ID); cs_handle_channel_join(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_WELCOME: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → handle WELCOME", CS_ID); cs_handle_welcome(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_PEER_UPSERT: DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → handle PEER_UPSERT", CS_ID); cs_handle_peer_upsert(g_cs, peer, ch_id, pl, plen); break;
case CS_MSG_ERROR: {
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: RECV ERROR from=%016llx ch=%s", CS_ID, (unsigned long long)peer, ch_id);
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: RECV ERROR from=%016llx ch=%s", CS_ID, (unsigned long long)peer, ch_id);
if (g_cs->info_req_timer) { uasync_cancel_timeout(g_cs->inst->ua, g_cs->info_req_timer); g_cs->info_req_timer = NULL; }
uint8_t err[20]; memcpy(err, &peer, 8); int r = CC_ERR_NOT_FOUND; memcpy(err + 8, &r, 4);
memcpy(err + 12, &g_cs->pending_invite_ch_id, 8); gui_bridge_post(GUI_EVT_CONNECT_RESULT, err, 20);
@ -394,7 +396,7 @@ static void chat_sync_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
break;
}
case CS_MSG_PEER_REMOVE: cs_handle_peer_remove(g_cs, peer, ch_id, pl, plen); break;
default: DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: UNKNOWN msg type=%02x from=%016llx", CS_ID, type, (unsigned long long)peer); break;
default: DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: UNKNOWN msg type=%02x from=%016llx", CS_ID, type, (unsigned long long)peer); break;
}
u_free(entry->dgram); queue_entry_free(entry);
}
@ -405,7 +407,7 @@ static void cs_info_req_timeout_cb(void* arg) {
struct chat_sync* cs = (struct chat_sync*)arg;
if (!cs || !cs->initialized) return;
cs->info_req_timer = NULL;
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_INFO_REQ timeout peer=%016llx ch=%llu",
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_REQ timeout peer=%016llx ch=%llu",
CS_ID, (unsigned long long)cs->pending_invite_node_id,
(unsigned long long)cs->pending_invite_ch_id);
uint8_t err[12]; memcpy(err, &cs->pending_invite_node_id, 8);
@ -420,7 +422,7 @@ static void cs_join_timeout_cb(void* arg) {
if (!cs || !cs->initialized) return;
cs->join_timer = NULL;
uint64_t peer = cs->pending_invite_node_id;
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_JOIN timeout peer=%016llx ch=%llu",
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_JOIN timeout peer=%016llx ch=%llu",
CS_ID, (unsigned long long)peer, (unsigned long long)cs->pending_invite_ch_id);
uint8_t err[20]; memcpy(err, &peer, 8);
int r = CC_ERR_TIMEOUT; memcpy(err + 8, &r, 4);
@ -433,16 +435,16 @@ static void cs_join_timeout_cb(void* arg) {
/* ── helper: find active ETCP_CONN for node ── */
static struct ETCP_CONN* cs_find_conn_for_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
if (!inst->connections) { DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: find_conn connections=NULL", CS_ID); return NULL; }
if (!inst->connections) { DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn connections=NULL", CS_ID); return NULL; }
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node_id);
if (e) {
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: find_conn node=%016llx found=%p initialized=%d links_up=%d state=%d",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn node=%016llx found=%p initialized=%d links_up=%d state=%d",
CS_ID, (unsigned long long)node_id, (void*)ce->conn, ce->conn->initialized, ce->conn->links_up, ce->conn->state);
if (ce->conn->initialized && ce->conn->links_up) return ce->conn;
return NULL;
}
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: find_conn node=%016llx NOT FOUND in queue (head=%p count=%d)",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn node=%016llx NOT FOUND in queue (head=%p count=%d)",
CS_ID, (unsigned long long)node_id, (void*)inst->connections->head, queue_entry_count(inst->connections));
return NULL;
}
@ -482,7 +484,7 @@ static void cs_post_channel_online(struct chat_sync* cs, const char* ch_id) {
static void _on_member_sync_done(uint64_t peer, const char* ns, int result, void* arg) {
struct channel_cache* ch = (struct channel_cache*)arg;
if (result == MT_OK && ch) ch->synced = CS_SYNC_DONE;
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_sync %s ns=%s peer=%016llx",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: member_sync %s ns=%s peer=%016llx",
CS_ID, result == MT_OK ? "OK" : "FAIL", ns, (unsigned long long)peer);
}
@ -491,15 +493,16 @@ static void cs_on_conn_up(struct ETCP_CONN* conn, void* arg) {
if (!conn || !g_cs) return;
uint64_t peer = conn->peer_node_id;
if (peer == 0 || peer == g_cs->inst->node_id) return;
if (!conn->initialized) { DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: conn_up SKIP — not initialized peer=%016llx", CS_ID, (unsigned long long)peer); return; }
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: conn_up peer=%016llx init=%d links=%d", CS_ID, (unsigned long long)peer, conn->initialized, conn->links_up);
if (!conn->initialized) { DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: conn_up SKIP — not initialized peer=%016llx", CS_ID, (unsigned long long)peer); return; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: conn_up peer=%016llx pending_invite=%llu links_up=%d", CS_ID, (unsigned long long)peer, (unsigned long long)g_cs->pending_invite_ch_id, conn->links_up);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: conn_up peer=%016llx pending_invite=%llu links_up=%d", CS_ID, (unsigned long long)peer, (unsigned long long)g_cs->pending_invite_ch_id, conn->links_up);
if (g_cs->pending_invite_ch_id != 0 && !g_cs->info_req_timer) {
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: conn_up invite path peer=%016llx ch=%llu",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: conn_up invite path peer=%016llx ch=%llu",
CS_ID, (unsigned long long)peer, (unsigned long long)g_cs->pending_invite_ch_id);
if (g_cs->pending_invite_node_id != 0 && g_cs->pending_invite_node_id != peer)
DEBUG_WARN(DEBUG_CATEGORY_DEBUG, "%s: invite node_id MISMATCH: invite=0x%016llx ETCP_peer=0x%016llx — invite is STALE!",
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: invite node_id MISMATCH: invite=0x%016llx ETCP_peer=0x%016llx — invite is STALE!",
CS_ID, (unsigned long long)g_cs->pending_invite_node_id, (unsigned long long)peer);
g_cs->pending_invite_node_id = peer;
char ch_id_str[64];
@ -523,17 +526,18 @@ static void cs_on_conn_down(struct ETCP_CONN* conn, void* arg) {
(void)arg;
if (!conn || !g_cs) return;
uint64_t peer = conn->peer_node_id;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: conn_down peer=%016llx", CS_ID, (unsigned long long)peer);
uint16_t rtt = conn->rtt_avg_100;
if (rtt > 0 && peer != 0 && g_cs->inst->topo_sqlite_db) {
sqlite3* db = g_cs->inst->topo_sqlite_db;
sqlite3_stmt* st = NULL;
sqlite3_prepare_v2(db, "UPDATE node_addresses SET rtt=? WHERE node_id=?", -1, &st, NULL);
if (st) { sqlite3_bind_int(st, 1, (int)rtt); sqlite3_bind_int64(st, 2, (sqlite3_int64)peer); sqlite3_step(st); sqlite3_finalize(st); }
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: conn_down saved rtt=%u for node %016llx", CS_ID, rtt, (unsigned long long)peer);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: conn_down saved rtt=%u for node %016llx", CS_ID, rtt, (unsigned long long)peer);
}
cs_cancel_proto_timers(g_cs);
if (g_cs->pending_invite_node_id == peer) {
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: conn down while waiting invite resp peer=%016llx, timers cancelled, state kept for retry on reconnect",
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: conn down while waiting invite resp peer=%016llx, timers cancelled, state kept for retry on reconnect",
CS_ID, (unsigned long long)peer);
/* invite state survives connection flaps: only timeout or explicit protocol end clears it */
}
@ -554,6 +558,7 @@ static void cs_on_conn_down(struct ETCP_CONN* conn, void* arg) {
static void cs_on_new_conn(struct ETCP_CONN* conn, void* arg) {
(void)arg;
if (!conn) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: new_conn conn=%p", CS_ID, (void*)conn);
etcp_conn_add_up_cbk(conn, cs_on_conn_up, NULL);
etcp_conn_add_down_cbk(conn, cs_on_conn_down, NULL);
}
@ -618,6 +623,7 @@ int chat_sync_init(struct UTUN_INSTANCE* inst,
void (*gui_cb)(void*, int, const uint8_t*, int)) {
(void)gui_cb;
if (!inst) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: init", CS_ID);
struct chat_sync* cs = u_calloc(1, sizeof(*cs));
if (!cs) return -1;
cs->inst = inst;
@ -646,13 +652,14 @@ int chat_sync_init(struct UTUN_INSTANCE* inst,
member_sync_init(inst);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: initialized", CS_ID);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: initialized", CS_ID);
return 0;
}
void chat_sync_destroy(struct UTUN_INSTANCE* inst) {
struct chat_sync* cs = g_cs;
if (!cs || !inst) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: destroy", CS_ID);
chat_sync_auto_connect_stop();
member_sync_destroy(inst);
cs->initialized = 0; g_cs = NULL;
@ -707,8 +714,10 @@ void chat_sync_connect_from_invite(uint64_t channel_id, uint64_t node_id,
g_cs->pending_invite_ch_id = channel_id;
g_cs->pending_invite_node_id = node_id;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: invite ch=%llu node=%016llx pubkey=%016llx addrs=%d",
CS_ID, channel_id, node_id, *(const uint64_t*)pubkey_bin, addr_count);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "%s: invite start ch=%llu node=0x%016llx pubkey=%016llx... addrs=%d",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: invite start ch=%llu node=0x%016llx pubkey=%016llx... addrs=%d",
CS_ID, channel_id, node_id, *(const uint64_t*)pubkey_bin, addr_count);
gui_bridge_post_uasync_fn(
@ -720,7 +729,7 @@ void chat_sync_connect_from_invite(uint64_t channel_id, uint64_t node_id,
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_CONNECTIVITY, "%s: EVP_PKEY_new failed", CS_ID); return -1; }
if (!pkey) { DEBUG_ERROR(DEBUG_CATEGORY_DB_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)
@ -733,7 +742,7 @@ static int cs_ed25519_sign(const uint8_t* privkey, const uint8_t* msg, size_t ms
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_CONNECTIVITY, "%s: EVP_PKEY_new pub failed", CS_ID); return -1; }
if (!pkey) { DEBUG_ERROR(DEBUG_CATEGORY_DB_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)
@ -779,16 +788,17 @@ static int _get_node_name(sqlite3* db, uint64_t node_id, char* out, size_t sz) {
static void cs_handle_channel_info_req(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
(void)pl; (void)len;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: INFO_REQ ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer);
char name[128]; int is_dm; uint64_t owner; uint8_t x25519[32], ed_pub[32], ch_sig[64];
if (topo_node_sqlite_channel_get(cs->inst->topo_sqlite_db,
ch_id, name, (int)sizeof(name), &is_dm, &owner, x25519, ed_pub, ch_sig) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_INFO_REQ unknown ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer);
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_REQ unknown ch=%s from=%016llx", CS_ID, ch_id, (unsigned long long)peer);
uint8_t err[1] = { CS_MSG_ERROR }; cs_send(cs, ch_id, peer, err, 1);
return;
}
uint64_t myid = cs->inst->node_id;
if (memcmp(cs->inst->my_keys.public_key, x25519, 32) != 0)
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_INFO_RESP pubkey MISMATCH: my=%016llx ch=%016llx — channel was created with DIFFERENT keys!",
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_RESP pubkey MISMATCH: my=%016llx ch=%016llx — channel was created with DIFFERENT keys!",
CS_ID, *(const uint64_t*)cs->inst->my_keys.public_key, *(const uint64_t*)x25519);
/* load or create join_sig */
uint8_t my_join_sig[64] = {0}; uint64_t my_join_ts = 0;
@ -802,7 +812,7 @@ static void cs_handle_channel_info_req(struct chat_sync* cs, uint64_t peer,
my_join_ts = (uint64_t)ntp_time_get_seconds(cs->inst);
memcpy(join_msg + mlen, &my_join_ts, 8); mlen += 8;
cs_ed25519_sign(cs->inst->my_ed25519_privkey, join_msg, mlen, my_join_sig);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: CHANNEL_INFO_RESP created NEW join_sig: myid=%016llx ts=%llu",
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_RESP created NEW join_sig: myid=%016llx ts=%llu",
CS_ID, (unsigned long long)myid, (unsigned long long)my_join_ts);
}
}
@ -846,10 +856,11 @@ static void cs_handle_channel_info_req(struct chat_sync* cs, uint64_t peer,
static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: INFO_RESP ch=%s from=%016llx len=%zu", CS_ID, ch_id, (unsigned long long)peer, len);
if (cs->info_req_timer) { uasync_cancel_timeout(cs->inst->ua, cs->info_req_timer); cs->info_req_timer = NULL; }
if (len < 1) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_INFO_RESP too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
if (len < 1) { DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_RESP too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
uint8_t nl = pl[0];
if (1 + nl + 8 + 1 + 32 + 32 + 64 + 1 + 64 + 8 + 1 > len) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_INFO_RESP truncated len=%zu", CS_ID, len); return; }
if (1 + nl + 8 + 1 + 32 + 32 + 64 + 1 + 64 + 8 + 1 > len) { DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_RESP truncated len=%zu", CS_ID, len); return; }
const uint8_t* p = pl + 1;
char name[128]; memcpy(name, p, nl); name[nl] = '\0'; p += nl;
uint64_t owner; memcpy(&owner, p, 8); p += 8;
@ -867,7 +878,7 @@ static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
char inv_name[128] = "";
if (p + inv_name_len <= pl + len) { memcpy(inv_name, p, inv_name_len); inv_name[inv_name_len] = '\0'; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: CHANNEL_INFO_RESP from peer=%016llx owner=%016llx ch_x25519=%016llx ch_ed=%016llx name=%s",
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_RESP from peer=%016llx owner=%016llx ch_x25519=%016llx ch_ed=%016llx name=%s",
CS_ID, (unsigned long long)peer, (unsigned long long)owner,
*(const uint64_t*)x25519, *(const uint64_t*)ed_pub, name);
@ -879,7 +890,7 @@ static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
memcpy(vmsg + vlen, x25519, 32); vlen += 32;
memcpy(vmsg + vlen, ed_pub, 32); vlen += 32;
if (cs_ed25519_verify(ed_pub, vmsg, vlen, ch_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_INFO_RESP invalid ch_sig ch=%s", CS_ID, ch_id);
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_RESP invalid ch_sig ch=%s", CS_ID, ch_id);
return;
}
@ -887,13 +898,13 @@ static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
topo_node_sqlite_channel_put(cs->inst->topo_sqlite_db,
ch_id, name, (int)is_dm, owner, x25519, NULL, ed_pub, NULL, ch_sig);
chat_core_ensure_channel_ready(ch_id);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: channel ready for sync ch=%s, db_sync will pick up via periodic check or active conn",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: channel ready for sync ch=%s, db_sync will pick up via periodic check or active conn",
CS_ID, ch_id);
/* verify inviter's join_sig AND save node info using ETCP-authenticated keys */
{ struct ETCP_CONN* inv_conn = cs_find_conn_for_node(cs->inst, peer);
if (!inv_conn) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_INFO_RESP no ETCP conn for inviter peer=%016llx", CS_ID, (unsigned long long)peer);
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_RESP no ETCP conn for inviter peer=%016llx", CS_ID, (unsigned long long)peer);
return;
}
const uint8_t* inv_x25519 = inv_conn->crypto_ctx.peer_public_key;
@ -908,7 +919,7 @@ static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
if (inviter_join_sig) memcpy(ivmsg + ilen, inviter_join_sig, 64); else memset(ivmsg + ilen, 0, 64);
ilen += 64;
if (cs_ed25519_verify(inv_ed, ivmsg, ilen, inviter_update_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_INFO_RESP invalid inviter_update_sig peer=%016llx ts=%llu",
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_INFO_RESP invalid inviter_update_sig peer=%016llx ts=%llu",
CS_ID, (unsigned long long)peer, (unsigned long long)inviter_update_ts);
}
/* save inviter node_info to local DB */
@ -936,7 +947,7 @@ static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
sqlite3_bind_blob(as, 2, ip, 4, SQLITE_STATIC);
sqlite3_bind_int(as, 3, port);
sqlite3_step(as); sqlite3_finalize(as);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: saved inviter addr peer=%016llx %d.%d.%d.%d:%d", CS_ID,
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: saved inviter addr peer=%016llx %d.%d.%d.%d:%d", CS_ID,
(unsigned long long)peer, ip[0], ip[1], ip[2], ip[3], port);
}
}
@ -944,7 +955,7 @@ static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
}
}
int rc = topo_node_sqlite_member_put(vdb, ch_id, peer, inviter_join_sig, inviter_join_ts, inviter_update_sig, inviter_update_ts, inv_x25519, inv_ed, inv_name, inviter_join_sig);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_put(inviter) FAILED ch=%s peer=%016llx rc=%d", CS_ID, ch_id, (unsigned long long)peer, rc);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_put(inviter) FAILED ch=%s peer=%016llx rc=%d", CS_ID, ch_id, (unsigned long long)peer, rc);
topo_node_sqlite_node_update_verified(vdb, peer, inv_name, inv_x25519, inv_ed, inviter_join_ts);
}
@ -1018,9 +1029,10 @@ static void cs_handle_channel_info_resp(struct chat_sync* cs, uint64_t peer,
static void cs_handle_channel_join(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
if (len < 8 + 32 + 32 + 64 + 8 + 1 + 1) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: CHANNEL_JOIN too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
if (len < 8 + 32 + 32 + 64 + 8 + 1 + 1) { DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: CHANNEL_JOIN too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
const uint8_t* p = pl;
uint64_t node_id; memcpy(&node_id, p, 8); p += 8;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: JOIN ch=%s node=%016llx from=%016llx", CS_ID, ch_id, (unsigned long long)node_id, (unsigned long long)peer);
const uint8_t* x25519 = p; p += 32;
const uint8_t* ed_pub = p; p += 32;
const uint8_t* join_sig = p; p += 64;
@ -1039,7 +1051,7 @@ static void cs_handle_channel_join(struct chat_sync* cs, uint64_t peer,
size_t nl = strlen(nm); memcpy(vmsg + vlen, nm, nl); vlen += nl; vmsg[vlen++] = '\0'; }
memcpy(vmsg + vlen, &join_ts, 8); vlen += 8;
if (cs_ed25519_verify(ed_pub, vmsg, vlen, join_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: JOIN invalid sig node=0x%016llx ch=%s — signed(node=%016llx x25519=%016llx ed=%016llx name=%s ts=%llu)", CS_ID,
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: JOIN invalid sig node=0x%016llx ch=%s — signed(node=%016llx x25519=%016llx ed=%016llx name=%s ts=%llu)", CS_ID,
(unsigned long long)node_id, ch_id,
(unsigned long long)node_id, *(const uint64_t*)x25519, *(const uint64_t*)ed_pub, joiner_name, (unsigned long long)join_ts);
return;
@ -1047,7 +1059,7 @@ static void cs_handle_channel_join(struct chat_sync* cs, uint64_t peer,
sqlite3* db = cs->inst->topo_sqlite_db;
int rc = topo_node_sqlite_member_put(db, ch_id, node_id, join_sig, join_ts, NULL, 0, x25519, ed_pub, joiner_name, NULL);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_put(joiner) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_put(joiner) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
topo_node_sqlite_node_update_verified(db, node_id, joiner_name, x25519, ed_pub, join_ts);
/* save joiner node_info to local DB */
if (db && joiner_name[0]) {
@ -1127,10 +1139,10 @@ static void cs_handle_channel_join(struct chat_sync* cs, uint64_t peer,
/* initiate member sync with the joiner (server side) */
member_sync_start(cs->inst, node_id, ch_id, NULL, NULL);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: JOIN starting member_sync with joiner node=0x%016llx ch=%s",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: JOIN starting member_sync with joiner node=0x%016llx ch=%s",
CS_ID, (unsigned long long)node_id, ch_id);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: JOIN accepted node=0x%016llx ch=%s addrs=%d",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: JOIN accepted node=0x%016llx ch=%s addrs=%d",
CS_ID, (unsigned long long)node_id, ch_id, addr_cnt);
}
@ -1140,11 +1152,12 @@ static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
if (cs->join_timer) { uasync_cancel_timeout(cs->inst->ua, cs->join_timer); cs->join_timer = NULL; }
(void)peer;
if (len < 2) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: WELCOME too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
if (len < 2) { DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: WELCOME too short len=%zu peer=%016llx", CS_ID, len, (unsigned long long)peer); return; }
sqlite3* db = cs->inst->topo_sqlite_db;
const uint8_t* p = pl;
uint16_t pc; memcpy(&pc, p, 2); p += 2;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: WELCOME ch=%s from=%016llx peers=%u", CS_ID, ch_id, (unsigned long long)peer, pc);
for (uint16_t i = 0; i < pc; i++) {
if ((size_t)(p - pl) + 8 + 32 + 32 + 64 + 8 + 1 + 1 > len) break;
uint64_t node_id; memcpy(&node_id, p, 8); p += 8;
@ -1171,7 +1184,7 @@ static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
if (ver) {
if (EVP_DigestVerifyInit(ver, NULL, NULL, NULL, pkey) != 1
|| EVP_DigestVerify(ver, join_sig, 64, vmsg, vlen) != 1)
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: WELCOME invalid join_sig node=0x%016llx", CS_ID, (unsigned long long)node_id);
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: WELCOME invalid join_sig node=0x%016llx", CS_ID, (unsigned long long)node_id);
EVP_MD_CTX_free(ver);
}
EVP_PKEY_free(pkey);
@ -1179,7 +1192,7 @@ static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
}
int rc = topo_node_sqlite_member_put(db, ch_id, node_id, join_sig, join_ts, NULL, 0, x25519, ed_pub, peer_name, NULL);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_put(welcome) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_put(welcome) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
topo_node_sqlite_node_update_verified(db, node_id, peer_name, x25519, ed_pub, join_ts);
for (uint8_t j = 0; j < ac; j++) {
@ -1216,9 +1229,9 @@ static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
const char* my_name = cs->inst->name[0] ? cs->inst->name : "";
int src = topo_node_sqlite_member_put(db, ch_id, myid, NULL, 0, NULL, 0, my_x25, my_ed, my_name, NULL);
if (src != 0)
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_put(self) FAILED in WELCOME ch=%s rc=%d", CS_ID, ch_id, src);
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_put(self) FAILED in WELCOME ch=%s rc=%d", CS_ID, ch_id, src);
else
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: WELCOME added self to peers ch=%s node=0x%016llx", CS_ID, ch_id, (unsigned long long)myid);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: WELCOME added self to peers ch=%s node=0x%016llx", CS_ID, ch_id, (unsigned long long)myid);
}
uint8_t evt[65]; uint8_t ch_id_len = (uint8_t)strlen(ch_id);
@ -1227,7 +1240,7 @@ static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
gui_bridge_post(GUI_EVT_MEMBERS_CHANGED, evt, 1 + ch_id_len);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: WELCOME processed ch=%s peers=%d — message sync triggered via db_sync (active conn or peer_check timer every 5s)",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: WELCOME processed ch=%s peers=%d — message sync triggered via db_sync (active conn or peer_check timer every 5s)",
CS_ID, ch_id, pc);
}
@ -1235,7 +1248,7 @@ static void cs_handle_welcome(struct chat_sync* cs, uint64_t peer,
static void cs_handle_peer_upsert(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
if (len < 8 + 32 + 32 + 64 + 8 + 1 + 1 + 1) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: PEER_UPSERT too short len=%zu", CS_ID, len); return; }
if (len < 8 + 32 + 32 + 64 + 8 + 1 + 1 + 1) { DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: PEER_UPSERT too short len=%zu", CS_ID, len); return; }
const uint8_t* p = pl;
uint64_t node_id; memcpy(&node_id, p, 8); p += 8;
const uint8_t* x25519 = p; p += 32;
@ -1256,7 +1269,7 @@ static void cs_handle_peer_upsert(struct chat_sync* cs, uint64_t peer,
size_t nl = strlen(nm); memcpy(vmsg + vlen, nm, nl); vlen += nl; vmsg[vlen++] = '\0'; }
memcpy(vmsg + vlen, &join_ts, 8); vlen += 8;
if (cs_ed25519_verify(ed_pub, vmsg, vlen, join_sig) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: PEER_UPSERT invalid sig node=0x%016llx", CS_ID,
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: PEER_UPSERT invalid sig node=0x%016llx", CS_ID,
(unsigned long long)node_id);
return;
}
@ -1264,7 +1277,7 @@ static void cs_handle_peer_upsert(struct chat_sync* cs, uint64_t peer,
sqlite3* db = cs->inst->topo_sqlite_db;
int rc = topo_node_sqlite_member_put(db, ch_id, node_id, join_sig, join_ts, NULL, 0, x25519, ed_pub, peer_name, NULL);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_put(peer_upsert) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
if (rc != 0) DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_put(peer_upsert) FAILED ch=%s node=0x%016llx rc=%d", CS_ID, ch_id, (unsigned long long)node_id, rc);
topo_node_sqlite_node_update_verified(db, node_id, peer_name, x25519, ed_pub, join_ts);
if (db && peer_name[0]) {
sqlite3_stmt* ns = NULL;
@ -1304,7 +1317,7 @@ static void cs_handle_peer_upsert(struct chat_sync* cs, uint64_t peer,
/* propagate to others (except sender and the subject node) */
cs_propagate(cs, ch_id, peer, pl, len);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: PEER_UPSERT node=0x%016llx ch=%s",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: PEER_UPSERT node=0x%016llx ch=%s",
CS_ID, (unsigned long long)node_id, ch_id);
}
@ -1312,7 +1325,7 @@ static void cs_handle_peer_upsert(struct chat_sync* cs, uint64_t peer,
static void cs_handle_peer_remove(struct chat_sync* cs, uint64_t peer,
const char* ch_id, const uint8_t* pl, size_t len) {
if (len < 8) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: PEER_REMOVE too short len=%zu", CS_ID, len); return; }
if (len < 8) { DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: PEER_REMOVE too short len=%zu", CS_ID, len); return; }
uint64_t node_id; memcpy(&node_id, pl, 8);
sqlite3* db = cs->inst->topo_sqlite_db;
@ -1323,6 +1336,6 @@ static void cs_handle_peer_remove(struct chat_sync* cs, uint64_t peer,
cs_propagate(cs, ch_id, peer, pl, len);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: PEER_REMOVE node=0x%016llx ch=%s",
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: PEER_REMOVE node=0x%016llx ch=%s",
CS_ID, (unsigned long long)node_id, ch_id);
}

15
tools/chatgui/transport/member_sync.c
Unescape View File

@ -91,6 +91,7 @@ static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level,
uint64_t prefix64, EVP_MD_CTX* sha_ctx) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
sqlite3* db = _db(inst); if (!db) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: bucket_hash ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix64);
char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));
int mask_shift = 63 - (int)level * 5;
@ -129,7 +130,8 @@ static int _member_get_items(void* ctx, const char* ns, uint8_t level,
uint64_t prefix, uint8_t prefix_bytes,
uint8_t* buf, size_t* len) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx;
sqlite3* db = _db(inst); if (!db || !buf || !len) return -1;
sqlite3* db = _db(inst); if (!db || !buf || !len) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: get_items ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix);
char peers_tbl[128]; _peers_table(ns, peers_tbl, sizeof(peers_tbl));
int mask_shift = 63 - (int)level * 5;
@ -217,6 +219,7 @@ static int _member_apply_items(void* ctx, const char* ns,
if (len < 2) return -1;
uint16_t count; memcpy(&count, data, 2);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: apply_items ns=%s count=%u", MS_ID, ns, count);
const uint8_t* mp = data + 2; size_t mrem = len - 2;
for (uint16_t i = 0; i < count && mrem >= 148; i++) {
@ -260,7 +263,7 @@ static int _member_apply_items(void* ctx, const char* ns,
int ok = (EVP_DigestVerifyInit(ver, NULL, NULL, NULL, pkey) == 1)
&& (EVP_DigestVerify(ver, usig, 64, vmsg, vlen) == 1);
if (!ok)
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_sync invalid update_sig node=0x%016llx ns=%s", MS_ID,
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_sync invalid update_sig node=0x%016llx ns=%s", MS_ID,
(unsigned long long)nid, ns);
EVP_MD_CTX_free(ver);
}
@ -284,9 +287,10 @@ static const struct merkle_sync_data_ops g_member_ops = {
int member_sync_init(struct UTUN_INSTANCE* inst) {
if (!inst) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: init", MS_ID);
int rc = merkle_sync_init(inst, 0x31, &g_member_ops, inst);
if (rc != 0) return rc;
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: initialized, merkle_rc=%d", MS_ID, rc);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: initialized, merkle_rc=%d", MS_ID, rc);
return 0;
}
@ -312,13 +316,14 @@ int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
const char* name,
const uint8_t* addrs_data, int addr_count) {
if (!inst || !ch_id) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: put ch=%s nid=%016llx name=%s ac=%d", MS_ID, ch_id, (unsigned long long)member_id, name ? name : "", addr_count);
sqlite3* db = _db(inst); if (!db) return -1;
int rc = topo_node_sqlite_member_put(db, ch_id, member_id,
join_sig, join_ts, update_sig, update_ts,
x25519, ed25519, name, NULL);
if (rc != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: member_put FAILED ch=%s id=0x%016llx name=%s rc=%d",
DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: member_put FAILED ch=%s id=0x%016llx name=%s rc=%d",
MS_ID, ch_id, (unsigned long long)member_id, name ? name : "", rc);
return -1;
}
@ -356,6 +361,7 @@ int member_sync_put(struct UTUN_INSTANCE* inst, const char* ch_id,
int member_sync_del(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t member_id) {
if (!inst || !ch_id) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: del ch=%s nid=%016llx", MS_ID, ch_id, (unsigned long long)member_id);
sqlite3* db = _db(inst); if (!db) return -1;
topo_node_sqlite_member_del(db, ch_id, member_id);
merkle_sync_recompute_path(inst, ch_id, member_id);
@ -377,6 +383,7 @@ int member_sync_count(struct UTUN_INSTANCE* inst, const char* ch_id) {
void member_sync_set_online(struct UTUN_INSTANCE* inst, uint64_t node_id, int online) {
if (!inst) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: set_online nid=%016llx online=%d", MS_ID, (unsigned long long)node_id, online);
sqlite3* db = _db(inst); if (!db) return;
topo_node_sqlite_node_set_online(db, node_id, online);
}

43
tools/chatgui/transport/merkle_sync.c
Unescape View File

@ -86,14 +86,16 @@ static void _ensure_table(struct merkle_sync* ms) {
static int _recompute_bucket(struct merkle_sync* ms, const char* ns,
uint8_t level, uint64_t prefix64) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: ns=%s L%d/P%016llx", MS_ID, ns, level, (unsigned long long)prefix64);
sqlite3* db = _db(ms->inst); if (!db) return -1;
EVP_MD_CTX* ctx = EVP_MD_CTX_new();
EVP_DigestInit_ex(ctx, EVP_sha256(), NULL);
int count = ms->ops->update_bucket_hash(ms->data_ctx, ns, level, prefix64, ctx);
if (count < 0) { EVP_MD_CTX_free(ctx); return -1; }
if (count < 0) { DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → error", MS_ID); EVP_MD_CTX_free(ctx); return -1; }
if (count == 0) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → empty, deleted", MS_ID);
EVP_MD_CTX_free(ctx);
sqlite3_stmt* ds = NULL;
sqlite3_prepare_v2(db,
@ -107,6 +109,7 @@ static int _recompute_bucket(struct merkle_sync* ms, const char* ns,
uint8_t hash[MT_HASH_SIZE];
EVP_DigestFinal_ex(ctx, hash, NULL);
EVP_MD_CTX_free(ctx);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → saved, count=%d", MS_ID, count);
sqlite3_stmt* is = NULL;
sqlite3_prepare_v2(db,
@ -126,6 +129,7 @@ static int _recompute_bucket(struct merkle_sync* ms, const char* ns,
void merkle_sync_recompute_path(struct UTUN_INSTANCE* inst, const char* ns, uint64_t key) {
struct merkle_sync* ms = g_merkle;
if (!ms || !ms->initialized) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: ns=%s key=%016llx", MS_ID, ns, (unsigned long long)key);
for (uint8_t level = 1; level <= MT_MAX_LEVEL; level++)
_recompute_bucket(ms, ns, level, merkle_sync_level_prefix(key, level));
}
@ -173,11 +177,11 @@ static int _get_level_hashes(struct merkle_sync* ms, const char* ns,
/* ── Send helpers ── */
static struct ETCP_CONN* ms_find_conn_for_node(struct UTUN_INSTANCE* inst, uint64_t node_id) {
if (!inst->connections) { DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: find_conn — inst->connections is NULL", MS_ID); return NULL; }
if (!inst->connections) { DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — inst->connections is NULL", MS_ID); return NULL; }
struct ll_entry* e = queue_find_data_by_index(inst->connections, (const uint8_t*)&node_id);
if (!e) { DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: find_conn — node=%016llx NOT FOUND in connections", MS_ID, (unsigned long long)node_id); return NULL; }
if (!e) { DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — node=%016llx NOT FOUND in connections", MS_ID, (unsigned long long)node_id); return NULL; }
struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: find_conn — node=%016llx found=%d initialized=%d links_up=%d", MS_ID, (unsigned long long)node_id, 1, ce->conn->initialized, ce->conn->links_up);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: find_conn — node=%016llx found=%d initialized=%d links_up=%d", MS_ID, (unsigned long long)node_id, 1, ce->conn->initialized, ce->conn->links_up);
if (ce->conn->initialized && ce->conn->links_up) return ce->conn;
return NULL;
}
@ -193,15 +197,16 @@ static int _send_msg(struct merkle_sync* ms, uint64_t peer,
if (!entry) { u_free(buf); return -1; }
entry->dgram = buf; entry->len = 1 + len;
struct ETCP_CONN* conn = ms_find_conn_for_node(ms->inst, peer);
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: no conn for node %016llx", MS_ID, (unsigned long long)peer); u_free(buf); queue_entry_free(entry); return -1; }
if (!conn) { DEBUG_ERROR(DEBUG_CATEGORY_DB_SYNC, "%s: no conn for node %016llx", MS_ID, (unsigned long long)peer); u_free(buf); queue_entry_free(entry); return -1; }
int r = etcp_send(conn, entry);
if (r != 0) { u_free(buf); queue_entry_free(entry); }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: send OK peer=%016llx len=%zu rc=%d", MS_ID, (unsigned long long)peer, len, r);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: send OK peer=%016llx len=%zu rc=%d", MS_ID, (unsigned long long)peer, len, r);
return r;
}
static int _send_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns,
uint8_t level, uint64_t prefix, uint8_t prefix_bytes, int is_data) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: send_hashes peer=%016llx ns=%s L%d P%016llx is_data=%d", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, is_data);
uint8_t ch_len = (uint8_t)strlen(ns);
size_t max_sz = 1 + 1 + ch_len + 1 + 1 + prefix_bytes + 1 + 4 + MT_BUCKETS * MT_HASH_SIZE + 2 + 65536;
uint8_t* buf = u_malloc(max_sz);
@ -224,7 +229,7 @@ static int _send_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns,
} else {
uint32_t bitmap; uint8_t hashes[MT_BUCKETS][MT_HASH_SIZE];
_get_level_hashes(ms, ns, level, prefix, prefix_bytes, &bitmap, hashes);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: send_hashes peer=%016llx ns=%s level=%d prefix=%016llx bitmap=%08x", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, bitmap);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: send_hashes peer=%016llx ns=%s level=%d prefix=%016llx bitmap=%08x", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, bitmap);
memcpy(p, &bitmap, 4); p += 4;
for (int i = 0; i < MT_BUCKETS; i++)
if (bitmap & (1u << i)) { memcpy(p, hashes[i], MT_HASH_SIZE); p += MT_HASH_SIZE; }
@ -236,6 +241,7 @@ static int _send_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns,
static int _send_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
struct bucket_entry* buckets, int count) {
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: send_batch peer=%016llx ns=%s count=%d", MS_ID, (unsigned long long)peer, ns, count);
uint8_t ch_len = (uint8_t)strlen(ns);
size_t max_sz = 1 + 1 + ch_len + 1 + (size_t)count * (1 + 1 + 8 + 1 + 4 + MT_BUCKETS * MT_HASH_SIZE + 65536);
uint8_t* buf = u_malloc(max_sz);
@ -290,16 +296,18 @@ static void _session_start_timer(struct ms_session* s);
static void _session_timeout_cb(void* arg) {
struct ms_session* s = (struct ms_session*)arg;
if (!s || !s->active || !g_merkle || !g_merkle->initialized) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: timeout peer=%016llx ns=%s retry=%d/3", MS_ID, (unsigned long long)s->peer, s->ns, s->retries);
s->timer = NULL;
s->retries++;
if (s->retries > 3) {
DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: sync timeout peer=%016llx ns=%s", MS_ID,
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: → max retries, done(ERR_TIMEOUT)", MS_ID);
DEBUG_WARN(DEBUG_CATEGORY_DB_SYNC, "%s: sync timeout peer=%016llx ns=%s", MS_ID,
(unsigned long long)s->peer, s->ns);
s->active = 0;
if (s->done_cb) s->done_cb(s->peer, s->ns, MT_ERR_TIMEOUT, s->cb_arg);
return;
}
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: retry %d peer=%016llx ns=%s", MS_ID,
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: retry %d peer=%016llx ns=%s", MS_ID,
s->retries, (unsigned long long)s->peer, s->ns);
_send_hashes(g_merkle, s->peer, s->ns, 1, 0, 1, 0);
_session_start_timer(s);
@ -333,6 +341,7 @@ static void _handle_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns
uint8_t is_data = pl[2 + pb];
const uint8_t* payload = pl + 3 + pb;
size_t paylen = plen - 3 - pb;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: handle_hashes peer=%016llx ns=%s L%d/P%016llx is_data=%d", MS_ID, (unsigned long long)peer, ns, level, (unsigned long long)prefix, is_data);
struct ms_session* s = _session_find(ms, peer, ns);
if (s && s->timer) { uasync_cancel_timeout(g_merkle->inst->ua, s->timer); s->timer = NULL; }
@ -369,7 +378,7 @@ static void _handle_hashes(struct merkle_sync* ms, uint64_t peer, const char* ns
}
}
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: handle_hashes peer=%016llx ns=%s level=%d remote_bm=%08x local_bm=%08x differs=%08x is_data=%d",
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: handle_hashes peer=%016llx ns=%s level=%d remote_bm=%08x local_bm=%08x differs=%08x is_data=%d",
MS_ID, (unsigned long long)peer, ns, level, remote_bm, local_bm, differs, is_data);
if (differs == 0) { _session_done(s, MT_OK); return; }
@ -421,8 +430,9 @@ static void _handle_request(struct merkle_sync* ms, uint64_t peer, const char* n
const uint8_t* pl, size_t plen) {
if (plen < 1) return;
uint8_t count = pl[0]; const uint8_t* bp = pl + 1; size_t off = 0;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: handle_request peer=%016llx ns=%s count=%d", MS_ID, (unsigned long long)peer, ns, count);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: handle_request peer=%016llx ns=%s count=%d", MS_ID, (unsigned long long)peer, ns, count);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: handle_request peer=%016llx ns=%s count=%d", MS_ID, (unsigned long long)peer, ns, count);
struct bucket_entry buckets[32]; int bc = 0;
for (uint8_t i = 0; i < count && bc < 32; i++) {
@ -447,8 +457,9 @@ static void _handle_batch(struct merkle_sync* ms, uint64_t peer, const char* ns,
const uint8_t* pl, size_t plen) {
if (plen < 1) return;
uint8_t count = pl[0]; const uint8_t* bp = pl + 1; size_t rem = plen - 1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: handle_batch peer=%016llx ns=%s count=%d len=%zu", MS_ID, (unsigned long long)peer, ns, count, plen - 1);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: handle_batch peer=%016llx ns=%s count=%d len=%zu", MS_ID, (unsigned long long)peer, ns, count, plen - 1);
DEBUG_DEBUG(DEBUG_CATEGORY_DB_SYNC, "%s: handle_batch peer=%016llx ns=%s count=%d len=%zu", MS_ID, (unsigned long long)peer, ns, count, plen - 1);
for (uint8_t i = 0; i < count && rem >= 3; i++) {
uint8_t lvl = bp[0]; uint8_t pb_i = bp[1]; rem -= 2; bp += 2;
@ -495,6 +506,7 @@ static void _recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (dlen < (size_t)(2 + ch_len + 1)) { u_free(entry->dgram); queue_entry_free(entry); return; }
char ns[64]; memcpy(ns, d + 2, ch_len); ns[ch_len] = '\0';
uint8_t type = d[2 + ch_len];
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: recv type=%02x from=%016llx ns=%s len=%zu", MS_ID, type, (unsigned long long)peer, ns, dlen);
const uint8_t* pl = d + 3 + ch_len;
size_t plen = dlen - 3 - ch_len;
@ -552,6 +564,7 @@ static void _bg_timer_cb(void* arg) {
if (!ch_id) continue;
int rc = merkle_sync_bg_check(ms->inst, ch_id);
if (rc < 0) continue;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: bg_check ns=%s → %s", MS_ID, ch_id, rc == 0 ? "consistent" : "RECALCULATED");
break;
}
sqlite3_finalize(cs);
@ -565,6 +578,7 @@ static void _bg_timer_cb(void* arg) {
int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id,
const struct merkle_sync_data_ops* ops, void* data_ctx) {
if (!inst || !ops) return -1;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: init svc=%02x", MS_ID, svc_id);
struct merkle_sync* ms = u_calloc(1, sizeof(*ms));
if (!ms) return -1;
ms->inst = inst; ms->svc_id = svc_id; ms->ops = ops; ms->data_ctx = data_ctx;
@ -578,7 +592,7 @@ int merkle_sync_init(struct UTUN_INSTANCE* inst, uint8_t svc_id,
ms->bg_timer = uasync_set_timeout(inst->ua, (uint32_t)(MS_BG_INTERVAL_MS * 10),
ms, _bg_timer_cb, "ms_bg");
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: initialized svc=%02x", MS_ID, svc_id);
DEBUG_INFO(DEBUG_CATEGORY_DB_SYNC, "%s: initialized svc=%02x", MS_ID, svc_id);
return 0;
}
@ -608,7 +622,7 @@ int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
_ensure_table(ms);
struct ms_session* s = _session_find(ms, peer, ns);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "%s: start peer=%016llx ns=%s new=%d initialized=%d", MS_ID, (unsigned long long)peer, ns, s ? 0 : 1, ms->initialized);
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: start peer=%016llx ns=%s new=%d", MS_ID, (unsigned long long)peer, ns, s ? 0 : 1);
if (!s) {
s = u_calloc(1, sizeof(*s));
if (!s) return -1;
@ -631,6 +645,7 @@ int merkle_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer,
void merkle_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ns) {
struct merkle_sync* ms = g_merkle;
if (!ms || !ns) return;
DEBUG_TRACE(DEBUG_CATEGORY_DB_SYNC, "%s: cancel peer=%016llx ns=%s", MS_ID, (unsigned long long)peer, ns);
struct ms_session** p = &ms->sessions;
while (*p) {
struct ms_session* s = *p;

Write Preview
Loading…
Cancel
Save