From 5e146f20bce214f2a35eaf554327faec4d96d47a Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 6 Oct 2026 19:56:52 +0200 Subject: [PATCH] Adapt member sync and protocol tests to dataset cursors --- src/chat/member_sync.c | 23 ++++++++++++++--------- tests/test_merkle_protocol.c | 30 +++++++++++++++--------------- 2 files changed, 29 insertions(+), 24 deletions(-) diff --git a/src/chat/member_sync.c b/src/chat/member_sync.c index 28f405a1..d4bf8799 100644 --- a/src/chat/member_sync.c +++ b/src/chat/member_sync.c @@ -204,12 +204,17 @@ static int _member_update_bucket_hash(void* ctx, const char* ns, uint8_t level, return count; } -static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int after_valid, uint64_t after, - uint8_t* buf, size_t* len, uint64_t* next, int* more) { +static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int after_valid, const merkle_key_t* after, + uint8_t* buf, size_t* len, merkle_key_t* next, int* more) { struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)ctx; sqlite3* db = _db(inst); if (!db || !buf || !len) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items — invalid args db=%p buf=%p len=%p", MS_ID, (void*)db, (void*)buf, (void*)len); return -1; } const uint8_t level = MT_MAX_LEVEL; - *next = 0; *more = 0; + memset(next, 0, sizeof(*next)); *more = 0; + uint64_t after_id = member_read64(after->bytes); + static const uint8_t zero_tail[MT_KEY_SIZE - 8]; + if (after_valid && memcmp(after->bytes + 8, zero_tail, sizeof(zero_tail))) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: invalid member cursor ns=%s", MS_ID, ns); return -1; + } DEBUG_TRACE(DEBUG_CATEGORY_MEMBER_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)); @@ -227,7 +232,7 @@ static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int afte sqlite3_stmt* stmt = NULL; if (sqlite3_prepare_v2(db, sql, -1, &stmt, NULL) != SQLITE_OK) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: get_items prepare failed ns=%s", MS_ID, ns); return -1; } sqlite3_bind_int(stmt, 1, after_valid); - sqlite3_bind_int64(stmt, 2, (sqlite3_int64)after); + sqlite3_bind_int64(stmt, 2, (sqlite3_int64)after_id); size_t off = 0; if (off + 2 > *len) { sqlite3_finalize(stmt); return -2; } @@ -290,7 +295,7 @@ static int _member_get_page(void* ctx, const char* ns, uint64_t prefix, int afte if (sgn && sqlite3_column_bytes(stmt, 11) >= 64) { memcpy(buf + off, sgn, 64); off += 64; } else { memset(buf + off, 0, 64); off += 64; } cnt++; - *next = nid; + member_write64(next->bytes, nid); } if (step != SQLITE_DONE && !*more) { DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "%s: page read failed ns=%s: %s", MS_ID, ns, sqlite3_errmsg(db)); sqlite3_finalize(stmt); return -1; } sqlite3_finalize(stmt); @@ -694,7 +699,7 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id, /* ── recompute и подтвердить реальное изменение по хешу ── */ if (changed) { - int rc = merkle_sync_recompute_path(inst, ch_id, m->node_id); + int rc = merkle_sync_recompute_path(inst, MT_DATA_MEMBERS, MT_DATA_MEMBERS, MT_DATA_MEMBERS, ch_id, m->node_id); if (rc < 0) return -1; if (rc == 0) changed = 0; /* запись была no-op (данные идентичны) */ } @@ -705,7 +710,7 @@ static int member_apply_record(struct UTUN_INSTANCE* inst, const char* ch_id, } static void member_committed(struct UTUN_INSTANCE* inst, const char* ns, uint64_t node_id) { - merkle_sync_changed(inst, ns); + merkle_sync_changed(inst, MT_DATA_MEMBERS, ns); struct TOPO_GROUP* g = inst->topo_groups ? topo_groups_find(inst->topo_groups, strtoull(ns, NULL, 10)) : NULL; struct ETCP_CONN* conn = instance_find_conn(inst, node_id); if (g && g->group_type == TOPO_GROUP_TYPE_CHAT) { @@ -1131,11 +1136,11 @@ void member_sync_destroy(struct UTUN_INSTANCE* inst) { int member_sync_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id, merkle_sync_done_cb done_cb, void* arg) { member_sync_verify_and_purge(inst, ch_id); - return merkle_sync_start(inst, peer, ch_id, done_cb, arg); + return merkle_sync_start(inst, MT_DATA_MEMBERS, peer, ch_id, done_cb, arg); } void member_sync_cancel(struct UTUN_INSTANCE* inst, uint64_t peer, const char* ch_id) { - merkle_sync_cancel(inst, peer, ch_id); + merkle_sync_cancel(inst, MT_DATA_MEMBERS, peer, ch_id); } void member_sync_cancel_channel(struct UTUN_INSTANCE* inst, const char* ch_id) { diff --git a/tests/test_merkle_protocol.c b/tests/test_merkle_protocol.c index 4329b71a..4b219d3b 100644 --- a/tests/test_merkle_protocol.c +++ b/tests/test_merkle_protocol.c @@ -14,10 +14,10 @@ static int t_hash(void* arg, const char* ns, uint8_t level, uint64_t prefix, EVP return EVP_DigestUpdate(hash, "record", 6) == 1 ? 1 : -1; } -static int t_page(void* arg, const char* ns, uint64_t prefix, int has_after, uint64_t after, - uint8_t* data, size_t* len, uint64_t* next, int* more) { +static int t_page(void* arg, const char* ns, uint64_t prefix, int has_after, const merkle_key_t* after, + uint8_t* data, size_t* len, merkle_key_t* next, int* more) { (void)arg; (void)ns; (void)prefix; (void)has_after; (void)after; - memset(data, 0, *len); *next = 1; *more = 1; return 0; + memset(data, 0, *len); memset(next, 0, sizeof(*next)); ms_write64(next->bytes, 1); *more = 1; return 0; } static int t_apply(void* arg, const char* ns, uint64_t peer, const uint8_t* data, size_t len) { @@ -66,7 +66,7 @@ static void t_destroy(struct fixture* f) { } static struct ms_session* t_session(struct fixture* f) { - struct ms_session* s = ms_create(f->inst.msync, &f->conn, 1001); assert(s); + struct ms_session* s = ms_create(f->inst.msync, &f->conn, MT_DATA_MEMBERS, 1001); assert(s); s->round = 7; s->state = MS_PULL; s->request_id = 1; s->waiting = 1; s->requests = u_calloc(1, sizeof(*s->requests)); assert(s->requests); s->requests->cb = t_done; s->requests->arg = f; @@ -75,7 +75,7 @@ static struct ms_session* t_session(struct fixture* f) { static void t_receive(struct fixture* f, uint8_t type, uint64_t round, uint32_t request, const uint8_t* data, size_t len) { struct ll_entry* e = ll_alloc_lldgram((uint16_t)(MS_HEADER + len)); assert(e); - e->len = (uint16_t)(MS_HEADER + len); e->dgram[0] = 0x72; + e->len = (uint16_t)(MS_HEADER + len); e->dgram[0] = 0x72; e->dgram[22] = MT_DATA_MEMBERS; ms_write64(e->dgram + 1, 1001); e->dgram[9] = type; ms_write64(e->dgram + 10, round); ms_write32(e->dgram + 18, request); if (len) memcpy(e->dgram + MS_HEADER, data, len); @@ -100,7 +100,7 @@ static void test_rejected_data(void) { struct fixture f; t_init(&f); struct ms_session* s = t_session(&f); s->stack[0].level = MT_MAX_LEVEL; f.apply_error = MT_ERR_DATA; - uint8_t page[11] = {0}; + uint8_t page[MS_PAGE_HEADER + 2] = {0}; t_receive(&f, MS_PAGE, 7, 1, page, sizeof(page)); assert(!f.callbacks && s->state == MS_FAILING); uasync_poll(f.inst.ua, 0); @@ -159,9 +159,9 @@ static void test_page_size_and_cursor(void) { static void test_readonly_tree(void) { struct fixture f; t_init(&f); assert(sqlite3_exec(f.inst.topo_sqlite_db, "PRAGMA query_only=ON", NULL, NULL, NULL) == SQLITE_OK); - assert(merkle_sync_recompute_path(&f.inst, "1001", 0) < 0); + assert(merkle_sync_recompute_path(&f.inst, MT_DATA_MEMBERS, "1001", 0) < 0); uint8_t hash[32], empty[32] = {0}; - assert(merkle_sync_read_hash(&f.inst, "1001", 0, 0, hash) == 0 && !memcmp(hash, empty, 32)); + assert(merkle_sync_read_hash(&f.inst, MT_DATA_MEMBERS, "1001", 0, 0, hash) == 0 && !memcmp(hash, empty, 32)); t_destroy(&f); puts("PASS: SQLite failure is reported and cannot publish a root"); } @@ -170,11 +170,11 @@ static void test_backpressure_and_disconnect(void) { struct fixture f; t_init(&f); struct ll_entry* blocker = queue_entry_new(0); assert(blocker); assert(queue_data_put(f.conn.send_input_q, blocker) == 0); - assert(merkle_sync_start(&f.inst, 2, "1001", t_done, &f) == 0); + assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_done, &f) == 0); uasync_poll(f.inst.ua, 0); struct ms_session* s = f.inst.msync->sessions; assert(s->tx && s->waiter.internal); - for (int i = 0; i < 1000; i++) merkle_sync_changed(&f.inst, "1001"); + for (int i = 0; i < 1000; i++) merkle_sync_changed(&f.inst, MT_DATA_MEMBERS, "1001"); assert(s->tx && queue_entry_count(f.conn.send_input_q) == 1 && s->dirty); f.conn.links_up = 0; ms_conn_status(&f.conn, ETCP_CONN_STATUS_DOWN, f.inst.msync); @@ -210,14 +210,14 @@ static void test_send_rejection(void) { static void t_cancel_done(uint64_t peer, const char* ns, int result, void* arg) { struct fixture* f = arg; t_done(peer, ns, result, arg); - merkle_sync_cancel(&f->inst, peer, ns); + merkle_sync_cancel(&f->inst, MT_DATA_MEMBERS, peer, ns); } static void test_error_handoff(void) { struct fixture f; t_init(&f); struct ms_session* s = t_session(&f); s->requests->cb = t_cancel_done; s->stack[0].level = MT_MAX_LEVEL; f.apply_error = MT_ERR_DATA; - uint8_t page[11] = {0}; + uint8_t page[MS_PAGE_HEADER + 2] = {0}; t_receive(&f, MS_PAGE, 7, 1, page, sizeof(page)); assert(!f.callbacks && s->state == MS_FAILING); uasync_poll(f.inst.ua, 0); @@ -249,15 +249,15 @@ static int t_begin_pull(void* ctx, const char* ns, uint64_t peer, uint64_t round } static int t_commit_pull(void* ctx, void* pull) { (void)ctx; assert(pull); pull_committed++; return 0; } static void t_abort_pull(void* ctx, void* pull) { (void)ctx; assert(pull); pull_released++; u_free(pull); } -static int t_stage_page(void* ctx, void* pull, uint64_t prefix, int av, uint64_t after, - uint64_t next, int more, const uint8_t* data, size_t len) { +static int t_stage_page(void* ctx, void* pull, uint64_t prefix, int av, const merkle_key_t* after, + const merkle_key_t* next, int more, const uint8_t* data, size_t len) { (void)ctx; (void)pull; (void)prefix; (void)av; (void)after; (void)next; (void)more; (void)data; (void)len; return 0; } static void test_pull_lifecycle(void) { struct fixture f; t_init(&f); struct merkle_sync_data_ops ops = t_ops; ops.begin_pull = t_begin_pull; ops.commit_pull = t_commit_pull; ops.abort_pull = t_abort_pull; ops.stage_page = t_stage_page; - f.inst.msync->ops = &ops; + f.inst.msync->datasets[MT_DATA_MEMBERS].ops = &ops; struct ms_session* s = t_session(&f); assert(ms_pull_start(s) == 0 && s->pull); ms_discard_tx(s); s->stack[0].level = MT_MAX_LEVEL;