From f9f516c2e9559d95a589ab235d17d8e9ea02bf22 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Tue, 14 Jul 2026 18:34:24 +0300 Subject: [PATCH] fix chatgui: PUSH new messages to peers + comprehensive debug logging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chat_sync_push: fix PUSH format to match cs_handle_push expectations (was missing insert_record header: ts+dh+nid before ct+dlen+data) Add DEBUG_INFO log with peer count after send. on_db_sync_insert: call chat_sync_push after successful local insert → new messages are now pushed to all connected peers Add DEBUG_ERROR to 4 silent returns (JSON parse fail, SQL errors) chat_core_submit_message: add DEBUG_INFO entry log (ch, ct, len) chat_core_insert_record: add DEBUG_WARN/ERROR to 7 silent returns (truncated record, SQL prepare fail, unknown step error) cs_handle_push: add DEBUG_INFO for ACK send, DEBUG_ERROR for insert fail cs_handle_send_data: DEBUG_WARN when record truncated mid-loop cs_on_conn_up: DEBUG_INFO sync path with channel count Full trace now shows every message: SUBMIT → insert → PUSH → RECV PUSH → ACK --- tools/chatgui/transport/chat_core.c | 37 +++++++++++++++++++---------- tools/chatgui/transport/chat_sync.c | 30 +++++++++++++++++++---- 2 files changed, 51 insertions(+), 16 deletions(-) diff --git a/tools/chatgui/transport/chat_core.c b/tools/chatgui/transport/chat_core.c index 807e67f0..a310e741 100644 --- a/tools/chatgui/transport/chat_core.c +++ b/tools/chatgui/transport/chat_core.c @@ -6,6 +6,7 @@ */ #include "chat_core.h" +#include "chat_sync.h" #include "db_sync.h" #include "gui_bridge.h" #include "topo_node_sqlite.h" @@ -312,6 +313,9 @@ void chat_core_submit_message(struct chat_msg_submit* req) { if (!g_cc.initialized) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: submit before init", CC_ID); return; } if (!req) return; + DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: SUBMIT ch=%s ct=%s len=%u", + CC_ID, req->channel_id, req->content_type, req->data_len); + struct DB_SYNC_INSTANCE* si = si_find(req->channel_id); if (!si) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: no db_sync instance for ch=%s", CC_ID, req->channel_id); return; } @@ -390,7 +394,7 @@ int chat_core_chain_hash_at(const char* ch_id, uint32_t pos, uint8_t* hash_out) } int chat_core_insert_record(const char* ch_id, const uint8_t* rec, size_t len) { - if (!g_cc.initialized || !rec || len < 29) return -1; + if (!g_cc.initialized || !rec || len < 29) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: insert_record malformed (len=%zu < 29) ch=%s", CC_ID, len, ch_id); return -1; } const uint8_t* p = rec; size_t rem = len; @@ -398,16 +402,16 @@ int chat_core_insert_record(const char* ch_id, const uint8_t* rec, size_t len) { memcpy(&ts, p, 8); p += 8; rem -= 8; memcpy(&dh, p, 8); p += 8; rem -= 8; memcpy(&nid, p, 8); p += 8; rem -= 8; - if (rem < 1) return -1; + if (rem < 1) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: insert_record truncated at ct_len ch=%s", CC_ID, ch_id); return -1; } uint8_t ct_len = *p; p++; rem--; - if (rem < ct_len) return -1; + if (rem < ct_len) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: insert_record truncated at ct ch=%s", CC_ID, ch_id); return -1; } char ct_buf[64]; memcpy(ct_buf, p, ct_len); ct_buf[ct_len] = '\0'; p += ct_len; rem -= ct_len; - if (rem < 4) return -1; + if (rem < 4) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: insert_record truncated at dlen ch=%s", CC_ID, ch_id); return -1; } uint32_t dlen; memcpy(&dlen, p, 4); p += 4; rem -= 4; - if (rem < dlen) return -1; - const uint8_t* data = p; p += dlen; rem -= dlen; - if (rem < 32) return -1; + if (rem < dlen) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: insert_record truncated at data ch=%s", CC_ID, ch_id); return -1; } + const uint8_t* rdata = p; p += dlen; rem -= dlen; + if (rem < 32) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: insert_record truncated at chain_hash ch=%s", CC_ID, ch_id); return -1; } uint8_t peer_ch[32]; memcpy(peer_ch, p, 32); @@ -429,10 +433,10 @@ int chat_core_insert_record(const char* ch_id, const uint8_t* rec, size_t len) { " VALUES(?,?,?,?,?,?,?,0,1)", tbl); sqlite3_stmt* stmt = NULL; - if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + if (sqlite3_prepare_v2(g_cc.db, sql, -1, &stmt, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: insert_record prepare failed ch=%s", CC_ID, ch_id); return -1; } sqlite3_bind_int64(stmt, 1, (sqlite3_int64)nid); sqlite3_bind_text(stmt, 2, ct_buf, ct_len, SQLITE_STATIC); - sqlite3_bind_blob(stmt, 3, data, (int)dlen, SQLITE_STATIC); + sqlite3_bind_blob(stmt, 3, rdata, (int)dlen, SQLITE_STATIC); sqlite3_bind_int64(stmt, 4, ts); sqlite3_bind_int64(stmt, 5, (sqlite3_int64)dh); sqlite3_bind_blob(stmt, 6, peer_ch, 32, SQLITE_STATIC); @@ -448,6 +452,8 @@ int chat_core_insert_record(const char* ch_id, const uint8_t* rec, size_t len) { return 0; } if (rc == SQLITE_CONSTRAINT) return 1; + DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: insert_record step failed rc=%d ch=%s ts=%lld dh=0x%016llx", + CC_ID, rc, ch_id, (long long)ts, (unsigned long long)dh); return -1; } @@ -1005,7 +1011,7 @@ static void on_db_sync_insert(struct DB_SYNC_INSTANCE* si, const char* data, siz /* parse JSON: {"n":,"ch":"","ct":"","d":""} */ const char* p = data; const char* end = data + len; uint64_t jn = 0; { /* "n" */ - const char* n = strstr(p, "\"n\":"); if (!n) return; + const char* n = strstr(p, "\"n\":"); if (!n) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: on_db_sync_insert JSON parse fail: no '\"n\"' field ch=%s", CC_ID, ch_id); return; } n += 4; jn = strtoull(n, NULL, 0); p = n; } const char* jct = "text/plain"; size_t jct_len = 10; @@ -1016,7 +1022,7 @@ static void on_db_sync_insert(struct DB_SYNC_INSTANCE* si, const char* data, siz { /* "d" */ const char* d = strstr(p, "\"d\":\""); if (d) { d += 5; const char* de = d; while (de < end) { if (*de == '"' && (de == d || *(de - 1) != '\\')) break; de++; } if (de < end) { jd_start = d; jd_len = (size_t)(de - d); } } } - if (!jd_start) return; + if (!jd_start) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: on_db_sync_insert JSON parse fail: no '\"d\"' field ch=%s", CC_ID, ch_id); return; } uint64_t dh; compute_datahash((const uint8_t*)jd_start, jd_len, &dh); uint64_t ts = get_time_us(); @@ -1027,7 +1033,7 @@ static void on_db_sync_insert(struct DB_SYNC_INSTANCE* si, const char* data, siz "INSERT OR IGNORE INTO \"%s\" (node_id,content_type,data,timestamp,datahash,chain_hash,signature,is_outgoing,is_read,sync_flags)" " VALUES(?,?,?,?,?,?,?,?,?,?)", tbl); sqlite3_stmt* st=NULL; - if (sqlite3_prepare_v2(g_cc.db,sql,-1,&st,NULL)!=SQLITE_OK) return; + if (sqlite3_prepare_v2(g_cc.db,sql,-1,&st,NULL)!=SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: on_db_sync_insert prepare failed ch=%s", CC_ID, ch_id); return; } sqlite3_bind_int64(st,1,(sqlite3_int64)jn); sqlite3_bind_text(st,2,jct,(int)jct_len,SQLITE_STATIC); sqlite3_bind_blob(st,3,jd_start,(int)jd_len,SQLITE_STATIC); @@ -1041,6 +1047,13 @@ static void on_db_sync_insert(struct DB_SYNC_INSTANCE* si, const char* data, siz if (rc==SQLITE_DONE) { uint8_t evt[65]; uint8_t cl=(uint8_t)strlen(ch_id); evt[0]=cl; memcpy(evt+1,ch_id,cl); gui_bridge_post(GUI_EVT_MSG_RECEIVED, evt, 1+cl); + chat_sync_push(g_cc.inst, ch_id, jn, jct, (const uint8_t*)jd_start, (uint32_t)jd_len, ts, dh); + } else if (rc == SQLITE_CONSTRAINT) { + DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: on_db_sync_insert duplicate ch=%s ts=%lld dh=0x%016llx", + CC_ID, ch_id, (long long)ts, (unsigned long long)dh); + } else { + DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: on_db_sync_insert step failed rc=%d ch=%s", + CC_ID, rc, ch_id); } } diff --git a/tools/chatgui/transport/chat_sync.c b/tools/chatgui/transport/chat_sync.c index 64d265ae..55401b90 100644 --- a/tools/chatgui/transport/chat_sync.c +++ b/tools/chatgui/transport/chat_sync.c @@ -236,7 +236,7 @@ static void cs_handle_send_data(struct chat_sync* cs, uint64_t peer, size_t remain = len - 6; for (uint16_t i = 0; i < count && remain > 0; i++) { - if (remain < 29) break; + if (remain < 29) { DEBUG_WARN(DEBUG_CATEGORY_CONNECTIVITY, "%s: SEND_DATA record truncated (remain=%zu < 29) i=%u/%u ch=%s", CS_ID, remain, i, count, ch_id); break; } const uint8_t* rec_start = ptr; ptr += 8 + 8 + 8; remain -= 24; /* ts, dh, nid */ if (remain < 1) break; @@ -281,6 +281,8 @@ static void cs_handle_push(struct chat_sync* cs, uint64_t peer, uint8_t ack[17]; ack[0] = CS_MSG_ACK_PUSH; memcpy(ack + 1, &dh, 8); memcpy(ack + 9, &ts, 8); cs_send(cs, ch_id, peer, ack, 17); + DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: RECV PUSH OK ch=%s ts=%lld dh=0x%016llx, sending ACK", + CS_ID, ch_id, (long long)ts, (unsigned long long)dh); uint8_t evt[65]; evt[0] = (uint8_t)strlen(ch_id); memcpy(evt + 1, ch_id, evt[0]); @@ -288,6 +290,9 @@ static void cs_handle_push(struct chat_sync* cs, uint64_t peer, struct channel_cache* ch = cs_find(cs, ch_id); if (ch) ch->msg_count++; + } else { + DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: PUSH insert_record failed rc=%d ch=%s ts=%lld dh=0x%016llx", + CS_ID, rc, ch_id, (long long)ts, (unsigned long long)dh); } } @@ -437,6 +442,8 @@ static void cs_on_conn_up(struct ETCP_CONN* conn, void* arg) { return; } + /* sync path: initiate INIT_SYNC for shared channels */ + int synced_cnt = 0; for (int i = 0; i < g_cs->channel_count; i++) { struct channel_cache* ch = &g_cs->channels[i]; int found = 0; @@ -444,6 +451,7 @@ static void cs_on_conn_up(struct ETCP_CONN* conn, void* arg) { if (ch->peer_ids[j] == peer) { found = 1; break; } if (!found) continue; ch->synced = CS_SYNC_IN_PROGRESS; + synced_cnt++; uint8_t msg[37]; msg[0] = CS_MSG_INIT_SYNC; chat_core_chain_hash_at(ch->channel_id, ch->msg_count > 0 ? ch->msg_count - 1 : 0, ch->last_chain_hash); @@ -452,6 +460,10 @@ static void cs_on_conn_up(struct ETCP_CONN* conn, void* arg) { cs_send(g_cs, ch->channel_id, peer, msg, 37); member_sync_start(g_cs->inst, peer, ch->channel_id, _on_member_sync_done, ch); } + if (synced_cnt > 0) { + DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: conn_up sync path peer=%016llx, syncing %d channels", + CS_ID, (unsigned long long)peer, synced_cnt); + } member_sync_set_online(g_cs->inst, peer, 1); } @@ -617,9 +629,12 @@ int chat_sync_push(struct UTUN_INSTANCE* inst, if (!g_cs || !g_cs->initialized) return -1; struct channel_cache* ch = cs_find(g_cs, ch_id); + /* build record in insert_record format: ts(8)+dh(8)+nid(8)+ct(1)+ct+dlen(4)+data+chain_hash(32) */ uint8_t ct_len = content_type ? (uint8_t)strlen(content_type) : 0; if (ct_len > 63) ct_len = 63; - size_t total = 1 + 8 + 8 + 8 + 1 + ct_len + 4 + data_len + 32; + size_t rec_len = 24 + 1 + ct_len + 4 + data_len + 32; + /* PUSH wrapper: CS_MSG_PUSH(1) + header_ts(8) + header_dh(8) + pad(8) + record */ + size_t total = 1 + 24 + rec_len; uint8_t* buf = u_malloc(total); if (!buf) return -1; @@ -627,24 +642,31 @@ int chat_sync_push(struct UTUN_INSTANCE* inst, *p++ = CS_MSG_PUSH; memcpy(p, ×tamp, 8); p += 8; memcpy(p, &datahash, 8); p += 8; + uint64_t z = 0; memcpy(p, &z, 8); p += 8; /* pad */ + /* record: ts + dh + nid + ct + dlen + data + chain_hash */ + memcpy(p, ×tamp, 8); p += 8; + memcpy(p, &datahash, 8); p += 8; memcpy(p, &node_id, 8); p += 8; *p++ = ct_len; if (ct_len) { memcpy(p, content_type, ct_len); p += ct_len; } uint32_t dlen = data_len; memcpy(p, &dlen, 4); p += 4; if (data_len) { memcpy(p, data, data_len); p += data_len; } - memset(p, 0, 32); + memset(p, 0, 32); p += 32; int sent = 0; uint64_t myid = g_cs->inst->node_id; if (ch) { for (int i = 0; i < ch->peer_count; i++) { if (ch->peer_ids[i] == node_id || ch->peer_ids[i] == myid) continue; - cs_send(g_cs, ch_id, ch->peer_ids[i], buf, (size_t)(p + 32 - buf)); + cs_send(g_cs, ch_id, ch->peer_ids[i], buf, (size_t)(p - buf)); sent++; } } u_free(buf); + DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: PUSH sent to %d peers ch=%s ts=%lld dh=0x%016llx", + CS_ID, sent, ch_id, (long long)timestamp, (unsigned long long)datahash); + (void)inst; return sent; }