Test cancelled Merkle rounds against the fresh responder challenge

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.
 

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;