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

Diagnose Merkle source profiles and identify both exchange endpoints

Browse Source
This commit is contained in:
evgeny committed 2026-10-26 18:27:27 +01:00
1 parent 669ad86e57
commit e00d1f5212
2 files changed
+31 -8
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/chat/merkle_sync.c
+12 -8
View File
@@ -231,8 +231,9 @@ static void ms_sent(struct ll_queue* q, void* arg) {
return;
}
s->sent_bytes += len;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: TX dataset=%u ns=%s peer=%016llx round=%llu type=%u bytes=%zu",
s->dataset->id, s->ns, (unsigned long long)s->peer, (unsigned long long)s->round, type, len);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: TX self=%016llx dataset=%u ns=%s peer=%016llx round=%llu type=%u bytes=%zu",
(unsigned long long)s->ms->inst->node_id, s->dataset->id, s->ns,
(unsigned long long)s->peer, (unsigned long long)s->round, type, len);
if (type == MS_ERROR) {
s->state = MS_FAILED;
ms_stop_deadline(s);
@@ -280,8 +281,9 @@ static int ms_send(struct ms_session* s, int type, uint32_t request, const uint8
}
static void ms_fail(struct ms_session* s, int error, const char* reason, int report) {
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: failed dataset=%u ns=%s peer=%016llx round=%llu error=%d reason=%s",
s->dataset->id, s->ns, (unsigned long long)s->peer, (unsigned long long)s->round, error, reason);
DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: failed self=%016llx dataset=%u ns=%s peer=%016llx round=%llu error=%d reason=%s",
(unsigned long long)s->ms->inst->node_id, s->dataset->id, s->ns,
(unsigned long long)s->peer, (unsigned long long)s->round, error, reason);
s->state = MS_FAILED; s->dirty = 0; s->waiting = 0; s->failure_error = error;
if (s->work) { uasync_call_soon_cancel(s->ms->inst->ua, s->work); s->work = NULL; }
ms_discard_tx(s);
@@ -363,8 +365,8 @@ static int ms_begin(struct ms_session* s) {
memset(s->accept_id, 0, sizeof(s->accept_id));
s->request_id = s->remote_request_id = 0;
s->state = MS_BEGIN_WAIT; s->dirty = 0; s->waiting = 0;
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: begin dataset=%u ns=%s peer=%016llx round=%llu",
s->dataset->id, s->ns, (unsigned long long)s->peer, (unsigned long long)s->round);
DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: begin self=%016llx dataset=%u ns=%s peer=%016llx round=%llu",
(unsigned long long)s->ms->inst->node_id, s->dataset->id, s->ns, (unsigned long long)s->peer, (unsigned long long)s->round);
return ms_send(s, MS_BEGIN, 0, NULL, 0);
}
@@ -550,8 +552,10 @@ static void ms_receive(struct ETCP_CONN* conn, struct ll_entry* entry) {
s->received_bytes += entry->len;
const uint8_t* p = d + MS_HEADER;
size_t len = entry->len - MS_HEADER;
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC, "merkle_sync: RX dataset=%u ns=%s peer=%016llx round=%llu type=%u request=%u bytes=%zu state=%d",
dataset, ns, (unsigned long long)s->peer, (unsigned long long)round, type, request, len, s->state);
DEBUG_DEBUG(DEBUG_CATEGORY_MEMBER_SYNC,
"merkle_sync: RX self=%016llx dataset=%u ns=%s peer=%016llx round=%llu type=%u request=%u bytes=%zu state=%d",
(unsigned long long)ms->inst->node_id, dataset, ns, (unsigned long long)s->peer, (unsigned long long)round,
type, request, len, s->state);
int rc = 0, notify = 0;
if (s->state == MS_FAILING) goto done;
if (type != MS_WAKE && type != MS_BEGIN &&
tests/test_stream_service.c
+19
View File
@@ -354,6 +354,24 @@ static void wait_state(unsigned receiver, unsigned member, uint64_t group, uint6
}
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Merkle stream state timeout receiver=%u member=%u group=%llu epoch=%llx",
receiver, member, (unsigned long long)group, (unsigned long long)epoch);
for (unsigned i = 0; i < 4; i++) {
char query[160]; sqlite3_stmt* st = NULL;
snprintf(query, sizeof(query), "SELECT update_ts,userinfo FROM peers_%llu WHERE node_id=?", (unsigned long long)group);
int rc = sqlite3_prepare_v2(nodes[i]->topo_sqlite_db, query, -1, &st, NULL);
if (rc == SQLITE_OK) rc = sqlite3_bind_int64(st, 1, (sqlite3_int64)nodes[member]->node_id);
if (rc == SQLITE_OK) rc = sqlite3_step(st);
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Merkle profile node=%u self=%016llx member=%u sql=%d version=%lld userinfo=%s",
i, (unsigned long long)nodes[i]->node_id, member, rc,
rc == SQLITE_ROW ? (long long)sqlite3_column_int64(st, 0) : -1,
rc == SQLITE_ROW ? (const char*)sqlite3_column_text(st, 1) : "unavailable");
sqlite3_finalize(st);
for (unsigned j = 0; j < 4; j++) if (i != j) {
struct ETCP_CONN* conn = instance_find_conn(nodes[i], nodes[j]->node_id);
if (conn) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Merkle link self=%u peer=%u up=%d sleep=%d READY=%d",
i, j, conn->links_up, conn->peer_sleep_phase,
topo_group_peer_ready(topo_groups_find(nodes[i]->topo_groups, group), nodes[j]->node_id));
}
}
CHECK(0);
}
@@ -2345,6 +2363,7 @@ int main(int argc, char **argv) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG);
debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_INFO);
if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG);
debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_WARN);
#ifdef SIGPIPE
CHECK(signal(SIGPIPE, SIG_IGN) != SIG_ERR);