Test dataset isolation and colliding cursor prefixes

tests/test_member_adapter.c

@@ -63,9 +63,9 @@ static void test_owner_sizes(struct UASYNC* ua) {
         assert(sc_ed25519_sign(owner_priv, msg, n, owner) == SC_OK);
         assert(member_sync_apply_record(&source, "1001", source.node_id, &rec) & MS_APPLY_CHANGED);
         assert(member_sync_verify_local_record(source.topo_sqlite_db, "1001", &rec) == 1);
-        size_t bytes = sizeof(page); uint64_t next; int more;
-        assert(_member_get_page(&source, "1001", merkle_sync_level_prefix(nid, 5), 0, 0, page, &bytes, &next, &more) == 0);
-        assert(next == nid && !more && bytes < MT_PAGE_SIZE);
+        size_t bytes = sizeof(page); merkle_key_t next, after = {{0}}; int more;
+        assert(_member_get_page(&source, "1001", merkle_sync_level_prefix(nid, 5), 0, &after, page, &bytes, &next, &more) == 0);
+        assert(member_read64(next.bytes) == nid && !more && bytes < MT_PAGE_SIZE);
         struct member_wire w; size_t used;
         assert(member_decode(page + 2, bytes - 2, &w, &used) == 0 && used == bytes - 2 && !strcmp(w.rec.adm_tags, tags));
         assert(!strcmp(w.rec.userinfo, userinfo));
@@ -144,17 +144,17 @@ int main(void) {
     put(&source, 1); put(&source, 2);
 
     uint8_t raw[2049], *page = raw + 1; /* проверяем невыровненный буфер */
-    size_t len = 300; uint64_t next; int more;
-    assert(_member_get_page(&source, "1001", 0, 0, 0, page, &len, &next, &more) == 0);
-    assert(page[0] == 0 && page[1] == 1 && next == 1 && more == 1 && len <= 300);
+    size_t len = 300; merkle_key_t next, after = {{0}}; int more;
+    assert(_member_get_page(&source, "1001", 0, 0, &after, page, &len, &next, &more) == 0);
+    assert(page[0] == 0 && page[1] == 1 && member_read64(next.bytes) == 1 && more == 1 && len <= 300);
     struct member_wire w; size_t used;
     assert(member_decode(page + 2, len - 2, &w, &used) == 0 && w.rec.node_id == 1 && used == len - 2);
     len = 300;
-    assert(_member_get_page(&source, "1001", 0, 1, next, page, &len, &next, &more) == 0 && next == 2 && !more);
+    assert(_member_get_page(&source, "1001", 0, 1, &next, page, &len, &next, &more) == 0 && member_read64(next.bytes) == 2 && !more);
     puts("PASS: bounded member pages, canonical integers and unaligned buffers");
 
     len = 2048;
-    assert(_member_get_page(&source, "1001", 0, 0, 0, page, &len, &next, &more) == 0 && page[1] == 2 && !more);
+    assert(_member_get_page(&source, "1001", 0, 0, &after, page, &len, &next, &more) == 0 && page[1] == 2 && !more);
     assert(_member_apply_items(&target, "1001", 1, page, len - 1) < 0 && member_sync_count(&target, "1001") == 0);
     sqlite3* db = target.topo_sqlite_db;
     assert(sqlite3_exec(db, "CREATE TRIGGER fail_second BEFORE INSERT ON peers_1001 WHEN NEW.node_id=2 "
@@ -171,7 +171,7 @@ int main(void) {
     assert(merkle_sync_recompute_path(&target, "1001", 1) < 0);
     root(&target, hash); assert(!memcmp(hash, expected, 32));
     len = 2048;
-    assert(_member_get_page(&target, "1001", 0, 0, 0, page, &len, &next, &more) < 0);
+    assert(_member_get_page(&target, "1001", 0, 0, &after, page, &len, &next, &more) < 0);
     puts("PASS: corrupt BLOB cannot be hashed or serialized");
 
     member_sync_destroy(&target); member_sync_destroy(&source);

tests/test_merkle_protocol.c

@@ -272,6 +272,67 @@ static void test_pull_lifecycle(void) {
     puts("PASS: pull commits once; restart and disconnect abort exactly their own context");
 }
 
+/* Сессии и корни двух моделей одного канала не влияют друг на друга. */
+static void test_dataset_isolation(void) {
+    struct fixture f; t_init(&f);
+    assert(merkle_sync_register(&f.inst, MT_DATA_FEEDBACK, &t_ops, &f) == 0);
+    assert(merkle_sync_register(&f.inst, MT_DATA_FEEDBACK, &t_ops, &f) < 0);
+    assert(merkle_sync_recompute_path(&f.inst, MT_DATA_MEMBERS, "1001", 0) == 1);
+    uint8_t member[32], feedback[32], empty[32] = {0};
+    assert(merkle_sync_read_hash(&f.inst, MT_DATA_MEMBERS, "1001", 0, 0, member) == 0);
+    assert(merkle_sync_read_hash(&f.inst, MT_DATA_FEEDBACK, "1001", 0, 0, feedback) == 0);
+    assert(memcmp(member, empty, 32) && !memcmp(feedback, empty, 32));
+    assert(merkle_sync_recompute_path(&f.inst, MT_DATA_FEEDBACK, "1001", 0) == 1);
+    struct ms_session* a = ms_create(f.inst.msync, &f.conn, MT_DATA_MEMBERS, 1001); assert(a);
+    struct ms_session* b = ms_create(f.inst.msync, &f.conn, MT_DATA_FEEDBACK, 1001); assert(b);
+    assert(a != b && ms_find(f.inst.msync, &f.conn, MT_DATA_MEMBERS, 1001) == a);
+    a->state = b->state = MS_SERVE;
+    merkle_sync_changed(&f.inst, MT_DATA_FEEDBACK, "1001");
+    assert(!a->dirty && !a->revision && b->dirty && b->revision == 1);
+    merkle_sync_cancel(&f.inst, MT_DATA_FEEDBACK, 2, "1001");
+    assert(f.inst.msync->sessions == a && !a->next);
+    merkle_sync_cancel_ns(&f.inst, "1001"); assert(!f.inst.msync->sessions);
+    t_destroy(&f);
+    puts("PASS: dataset roots, sessions, changes and cancellation are independent");
+}
+
+/* Два разных полных ключа с одинаковыми первыми 64 битами проходят пагинацию. */
+static int t_full_page(void* ctx, const char* ns, uint64_t prefix, int av, const merkle_key_t* after,
+        uint8_t* data, size_t* len, merkle_key_t* next, int* more) {
+    (void)ctx; (void)ns; (void)prefix;
+    assert(*len >= 1);
+    *next = *after; ms_write64(next->bytes, 1);
+    next->bytes[MT_KEY_SIZE - 1] = av ? 2 : 1;
+    data[0] = next->bytes[MT_KEY_SIZE - 1]; *len = 1; *more = !av;
+    return 0;
+}
+
+static void test_full_cursor(void) {
+    struct fixture f; t_init(&f);
+    struct merkle_sync_data_ops ops = t_ops; ops.get_page = t_full_page;
+    assert(merkle_sync_register(&f.inst, MT_DATA_FEEDBACK, &ops, &f) == 0);
+    struct ms_session* s = ms_create(f.inst.msync, &f.conn, MT_DATA_FEEDBACK, 1001); assert(s);
+    s->state = MS_SERVE;
+    uint8_t query[MS_QUERY_SIZE] = { MT_MAX_LEVEL };
+    assert(ms_serve_query(s, 1, query, sizeof(query)) == 0);
+    assert(s->tx->dgram[22] == MT_DATA_FEEDBACK && s->tx->dgram[MS_HEADER] == 1);
+    memcpy(query + 10, s->tx->dgram + MS_HEADER + 1, MT_KEY_SIZE); query[9] = 1;
+    ms_discard_tx(s);
+    assert(ms_serve_query(s, 2, query, sizeof(query)) == 0);
+    assert(s->tx->dgram[MS_HEADER] == 0 && s->tx->dgram[MS_HEADER + MT_KEY_SIZE] == 2);
+    uint8_t page[MS_PAGE_HEADER + 1]; memcpy(page, s->tx->dgram + MS_HEADER, sizeof(page));
+    ms_discard_tx(s);
+    s->state = MS_PULL; s->stack[0].level = MT_MAX_LEVEL; s->waiting = 1; s->request_id = 1;
+    memcpy(s->after.bytes, query + 10, MT_KEY_SIZE); s->after_valid = 1;
+    page[0] = 1;
+    assert(ms_receive_page(s, 1, page, sizeof(page)) == 0);
+    assert(s->after.bytes[MT_KEY_SIZE - 1] == 2 && s->request_id == 2);
+    ms_discard_tx(s);
+    assert(ms_receive_page(s, 2, page, sizeof(page)) == MT_ERR_PROTOCOL);
+    t_destroy(&f);
+    puts("PASS: full cursors preserve colliding 64-bit prefixes and reject backwards progress");
+}
+
 int main(void) {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
     size_t baseline = u_get_allocated_count();
@@ -279,6 +340,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();
     assert(u_get_allocated_count() == baseline);
     puts("ALL PASS: protocol regressions, no tracked allocations leaked");
     return 0;