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

fix chatgui: PUSH new messages to peers + comprehensive debug logging 

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
topo_upd
Evgeny 3 months ago
parent
58b9f28149
commit
f9f516c2e9
2 changed files with 51 additions and 16 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. 37
      tools/chatgui/transport/chat_core.c
  2. 30
      tools/chatgui/transport/chat_sync.c

37
tools/chatgui/transport/chat_core.c
Unescape View File

@ -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":<uint64>,"ch":"<str>","ct":"<str>","d":"<str>"} */
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);
}
}

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

@ -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, &timestamp, 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, &timestamp, 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;
}

Write Preview
Loading…
Cancel
Save