From a079381c0a836c98623948d54756d4ebb9365fd0 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 16:36:13 +0100 Subject: [PATCH] Test cancelled Merkle rounds against the fresh responder challenge --- src/chat/merkle_sync_doc.md | 6 ++++- tests/test_merkle_protocol.c | 51 +++++++++++++++++++++++++++++++++--- 2 files changed, 52 insertions(+), 5 deletions(-) diff --git a/src/chat/merkle_sync_doc.md b/src/chat/merkle_sync_doc.md index 6bb42ea4..394e3f10 100644 --- a/src/chat/merkle_sync_doc.md +++ b/src/chat/merkle_sync_doc.md @@ -97,7 +97,11 @@ WAKE также перезапускает незавершённый раунд ## Wire -Заголовок: `service:1 | namespace:8 | type:1 | round:8 | request:4 | dataset:1` (23 байта). +Заголовок v4: `service:1 | namespace:8 | type:1 | round:8 | request:4 | dataset:1 | round_id:16 | accept_id:16` (55 байт). +`round` — диагностический счётчик. Свежий `round_id` создаёт инициатор BEGIN, свежий `accept_id` добавляет ответчик READY. +Все дальнейшие пакеты проверяют оба идентификатора, номер запроса и фазу. После cancel/start старый BEGIN может получить +новый challenge, но прежние PAGE/DONE/ERROR не соответствуют новому accept_id. Дубликат BEGIN не меняет текущий challenge. +WAKE содержит нулевые идентификаторы, BEGIN — нулевой accept_id. Старый wire не поддерживается. Все целые big-endian. Максимальный пакет вместе с заголовком — 16384 байта. Типы 0x10–0x19: WAKE, BEGIN, READY, QUERY, HASHES, PAGE, TURN, CHECK, DONE, ERROR. diff --git a/tests/test_merkle_protocol.c b/tests/test_merkle_protocol.c index b236f70d..189a07ec 100644 --- a/tests/test_merkle_protocol.c +++ b/tests/test_merkle_protocol.c @@ -6,7 +6,7 @@ struct fixture { struct UTUN_INSTANCE inst; struct ETCP_CONN conn; - int callbacks, result, apply_error, authorized; + int callbacks, result, apply_error, authorized, applied; }; static int t_hash(void* arg, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) { @@ -22,7 +22,8 @@ static int t_page(void* arg, const char* ns, uint64_t prefix, int has_after, con static int t_apply(void* arg, const char* ns, uint64_t peer, const uint8_t* data, size_t len) { (void)ns; (void)peer; (void)data; (void)len; - return ((struct fixture*)arg)->apply_error; + struct fixture* f = arg; + f->applied++; return f->apply_error; } static const struct merkle_sync_data_ops t_ops = { @@ -68,20 +69,62 @@ 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, MT_DATA_MEMBERS, 1001); assert(s); s->round = 7; s->state = MS_PULL; s->request_id = 1; s->waiting = 1; + ms_write64(s->round_id, 7); memset(s->accept_id, 0xa7, sizeof(s->accept_id)); s->requests = u_calloc(1, sizeof(*s->requests)); assert(s->requests); s->requests->cb = t_done; s->requests->arg = f; return s; } -static void t_receive(struct fixture* f, uint8_t type, uint64_t round, uint32_t request, const uint8_t* data, size_t len) { +static void t_receive_tokens(struct fixture* f, uint8_t type, uint64_t round, uint32_t request, + const uint8_t* round_id, const uint8_t* accept_id, 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->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); + memset(e->dgram + 23, 0, 2 * MS_TOKEN_SIZE); + if (round_id) memcpy(e->dgram + 23, round_id, MS_TOKEN_SIZE); + if (accept_id) memcpy(e->dgram + 39, accept_id, MS_TOKEN_SIZE); if (len) memcpy(e->dgram + MS_HEADER, data, len); ms_receive(&f->conn, e); } +static void t_receive(struct fixture* f, uint8_t type, uint64_t round, uint32_t request, const uint8_t* data, size_t len) { + struct ms_session* s = ms_find(f->inst.msync, &f->conn, MT_DATA_MEMBERS, 1001); + uint8_t id[MS_TOKEN_SIZE] = {0}; ms_write64(id, round); + t_receive_tokens(f, type, round, request, + type == MS_WAKE ? NULL : s && round == s->round ? s->round_id : id, + type == MS_WAKE || type == MS_BEGIN ? NULL : s ? s->accept_id : NULL, data, len); +} + +/* Старый BEGIN может получить новый challenge, но прежние PAGE/DONE/ERROR ему не соответствуют. */ +static void test_reattach_tokens(void) { + struct fixture f; t_init(&f); f.inst.node_id = 3; + struct ms_session* s = t_session(&f); + uint8_t old_round[MS_TOKEN_SIZE], old_accept[MS_TOKEN_SIZE]; + memcpy(old_round, s->round_id, sizeof(old_round)); memcpy(old_accept, s->accept_id, sizeof(old_accept)); + merkle_sync_cancel(&f.inst, MT_DATA_MEMBERS, 2, "1001"); + assert(merkle_sync_start(&f.inst, MT_DATA_MEMBERS, 2, "1001", t_done, &f) == 0); + s = f.inst.msync->sessions; + t_receive_tokens(&f, MS_BEGIN, 7, 0, old_round, NULL, NULL, 0); + assert(s->state == MS_SERVE && memcmp(old_accept, s->accept_id, sizeof(old_accept)) && s->tx); + uint8_t current_accept[MS_TOKEN_SIZE]; memcpy(current_accept, s->accept_id, sizeof(current_accept)); + t_receive_tokens(&f, MS_BEGIN, 7, 0, old_round, NULL, NULL, 0); + assert(!memcmp(current_accept, s->accept_id, sizeof(current_accept))); + ms_discard_tx(s); s->state = MS_PULL; s->waiting = 1; s->request_id = 1; s->stack[0].level = MT_MAX_LEVEL; + uint8_t page[MS_PAGE_HEADER + 1] = {0}, root[MT_HASH_SIZE] = {0}, error[4]; ms_write32(error, 3); + t_receive_tokens(&f, MS_PAGE, 7, 1, old_round, old_accept, page, sizeof(page)); + t_receive_tokens(&f, MS_DONE, 7, 0, old_round, old_accept, root, sizeof(root)); + t_receive_tokens(&f, MS_ERROR, 7, 0, old_round, old_accept, error, sizeof(error)); + assert(!f.applied && !f.callbacks && s->waiting && s->state == MS_PULL); + t_receive_tokens(&f, MS_PAGE, 7, 1, old_round, current_accept, page, sizeof(page)); + assert(f.applied == 1 && s->state == MS_DONE_WAIT && !f.callbacks); + ms_discard_tx(s); + t_receive_tokens(&f, MS_DONE, 7, 0, old_round, current_accept, root, sizeof(root)); + assert(f.callbacks == 1 && f.result == MT_OK); + t_destroy(&f); + DEBUG_INFO(DEBUG_CATEGORY_MEMBER_SYNC, "PASS fresh responder token fences old pages, completion and errors after cancel/start"); +} + static void test_malformed_and_old_round(void) { struct fixture f; t_init(&f); struct ms_session* s = t_session(&f); @@ -378,7 +421,7 @@ int main(void) { test_page_size_and_cursor(); test_readonly_tree(); test_backpressure_and_disconnect(); test_peer_restart(); test_send_rejection(); test_error_handoff(); test_blocked_error_timeout(); test_pull_lifecycle(); - test_dataset_isolation(); test_full_cursor(); test_permission_rounds(); + test_dataset_isolation(); test_full_cursor(); test_permission_rounds(); test_reattach_tokens(); assert(u_get_allocated_count() == baseline); puts("ALL PASS: protocol regressions, no tracked allocations leaked"); return 0;