Refactor Merkle sync into bounded bidirectional rounds with atomic member updates

tests/Makefile.am

@@ -66,6 +66,8 @@ check_PROGRAMS = \
 	test_node_conn_direct \
 	test_db_sync \
 	test_merkle_sync \
+	test_merkle_protocol \
+	test_member_adapter \
 	test_chat_sync_stress \
 	test_chat_join \
 	test_chat_join_e2e \
@@ -390,6 +392,12 @@ test_db_sync_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
 test_db_sync_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
 
 test_merkle_sync_SOURCES = test_merkle_sync.c
+test_merkle_protocol_SOURCES = test_merkle_protocol.c
+test_merkle_protocol_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib
+test_merkle_protocol_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
+test_member_adapter_SOURCES = test_member_adapter.c
+test_member_adapter_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib
+test_member_adapter_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
 test_merkle_sync_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib
 test_merkle_sync_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
 

tests/test_chat_join_e2e.c

@@ -556,10 +556,18 @@ static int child_main(const char* role, const char* dir, int invalid_key) {
             if (!wait_join_result(inst, 8000)) {
                 fprintf(stderr, "J: no JOIN_READY\n"); goto out;
             }
+            char ready[512]; snprintf(ready, sizeof(ready), "%s/join_ready", dir);
+            wf(ready, "ready");
             rc = 0;
         }
     }
 
+    /* Локальная запись не означает доставку: участники обслуживают обмен до JOIN_READY у J. */
+    if (rc == 0 && !invalid_key && strcmp(role, "j") != 0) {
+        char ready[512]; snprintf(ready, sizeof(ready), "%s/join_ready", dir);
+        if (!wait_file(inst, ready, 8000)) { fprintf(stderr, "%s: join delivery not confirmed\n", role); rc = 1; }
+    }
+
 out:
     if (early_link || delayed_registration) rc = 1;
     fprintf(stderr, "%s: %s\n", role, rc == 0 ? "OK" : "FAIL");

tests/test_media_delivery_full.c

@@ -461,6 +461,7 @@ int main(void) {
         g_inst[i] = utun_instance_create(g_ua, g_cfg[i]);
         if (!g_inst[i]) { printf("FAIL: create n%d\n", i); goto done; }
         utun_instance_init(g_inst[i]);
+        if (member_sync_init(g_inst[i]) < 0) { FAIL("member_sync_init n%d", i); goto done; }
     }
 
     mon(NULL);
@@ -491,7 +492,12 @@ int main(void) {
 
 done:
     g_result = 1;
-    for (int i = 0; i < N_NODES; i++) { if (g_inst[i]) { g_inst[i]->running = 0; utun_instance_destroy(g_inst[i]); g_inst[i] = NULL; } }
+    for (int i = 0; i < N_NODES; i++) {
+        if (g_inst[i]) {
+            member_sync_destroy(g_inst[i]);
+            g_inst[i]->running = 0; utun_instance_destroy(g_inst[i]); g_inst[i] = NULL;
+        }
+    }
     if (g_ua) { uasync_destroy(g_ua, 0); g_ua = NULL; }
     t_rmdir(g_tdir);
     return G_FAILED > 0 ? 1 : 0;

tests/test_member_adapter.c

@@ -0,0 +1,68 @@
+/* Реальный адаптер, SQLite и транзакции; без копии алгоритма и без сетевых таймингов. */
+#include "../src/chat/member_sync.c"
+#include <assert.h>
+
+static void setup(struct UTUN_INSTANCE* inst, struct UASYNC* ua) {
+    memset(inst, 0, sizeof(*inst)); inst->ua = ua; inst->node_id = 1;
+    assert(sqlite3_open(":memory:", &inst->topo_sqlite_db) == SQLITE_OK);
+    assert(topo_node_sqlite_init(inst->topo_sqlite_db) == 0);
+    uint8_t key[32] = {1}, signature[64] = {0};
+    assert(topo_node_sqlite_channel_put(inst->topo_sqlite_db, "1001", "test", 1, key, NULL, key, NULL, signature) == 0);
+    assert(member_sync_init(inst) == 0);
+}
+
+static void put(struct UTUN_INSTANCE* inst, uint64_t id) {
+    uint8_t key[32] = {1};
+    assert(member_sync_put(inst, "1001", id, key, key, NULL, 0, NULL, 0, "{}", NULL, NULL, 0, 0, NULL) >= 0);
+}
+
+static void root(struct UTUN_INSTANCE* inst, uint8_t hash[32]) {
+    assert(merkle_sync_read_hash(inst, "1001", 0, 0, hash) == 0);
+}
+
+int main(void) {
+    debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
+    size_t baseline = u_get_allocated_count();
+    struct UASYNC* ua = uasync_create(); assert(ua);
+    struct UTUN_INSTANCE source, target;
+    setup(&source, ua); setup(&target, ua);
+    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);
+    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);
+    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_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 "
+        "BEGIN SELECT RAISE(ABORT,'injected second record failure'); END", NULL, NULL, NULL) == SQLITE_OK);
+    assert(_member_apply_items(&target, "1001", 1, page, len) < 0);
+    uint8_t hash[32], empty[32] = {0}; root(&target, hash);
+    assert(member_sync_count(&target, "1001") == 0 && !memcmp(hash, empty, 32) && sqlite3_get_autocommit(db));
+    assert(sqlite3_exec(db, "DROP TRIGGER fail_second", NULL, NULL, NULL) == SQLITE_OK);
+    assert(_member_apply_items(&target, "1001", 1, page, len) == 0 && member_sync_count(&target, "1001") == 2);
+    uint8_t expected[32]; root(&source, expected); root(&target, hash); assert(!memcmp(hash, expected, 32));
+    puts("PASS: malformed page rejected; second-record error rolls back data and tree; complete page converges");
+
+    assert(sqlite3_exec(db, "UPDATE peers_1001 SET x25519_pubkey=X'01' WHERE node_id=1", NULL, NULL, NULL) == SQLITE_OK);
+    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);
+    puts("PASS: corrupt BLOB cannot be hashed or serialized");
+
+    member_sync_destroy(&target); member_sync_destroy(&source);
+    assert(sqlite3_close(target.topo_sqlite_db) == SQLITE_OK && sqlite3_close(source.topo_sqlite_db) == SQLITE_OK);
+    uasync_poll(ua, 0); uasync_destroy(ua, 0);
+    assert(u_get_allocated_count() == baseline);
+    puts("ALL PASS: member adapter");
+    return 0;
+}

tests/test_merkle_protocol.c

@@ -0,0 +1,215 @@
+/* Проверяем границы автомата на точных wire-пакетах. Включение реализации позволяет
+ * задать состояние без копирования private-структур и без подмены очередей/таймеров. */
+#include "../src/chat/merkle_sync.c"
+#include <assert.h>
+
+struct fixture {
+    struct UTUN_INSTANCE inst;
+    struct ETCP_CONN conn;
+    int callbacks, result, apply_error;
+};
+
+static int t_hash(void* arg, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* hash) {
+    (void)arg; (void)ns; (void)level; (void)prefix;
+    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) {
+    (void)arg; (void)ns; (void)prefix; (void)has_after; (void)after;
+    memset(data, 0, *len); *next = 1; *more = 1; return 0;
+}
+
+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;
+}
+
+static const struct merkle_sync_data_ops t_ops = {
+    .update_bucket_hash = t_hash, .get_page = t_page, .apply_items = t_apply
+};
+
+static void t_done(uint64_t peer, const char* ns, int result, void* arg) {
+    (void)peer; (void)ns;
+    struct fixture* f = arg;
+    f->callbacks++; f->result = result;
+}
+
+static void t_init(struct fixture* f) {
+    memset(f, 0, sizeof(*f));
+    f->inst.node_id = 1;
+    f->inst.ua = uasync_create(); assert(f->inst.ua);
+    assert(sqlite3_open(":memory:", &f->inst.topo_sqlite_db) == SQLITE_OK);
+    f->inst.connections = queue_new(f->inst.ua, 16, 0, 8, "test connections"); assert(f->inst.connections);
+    f->conn.instance = &f->inst; f->conn.peer_node_id = 2; f->conn.initialized = 1; f->conn.links_up = 1;
+    f->conn.send_input_q = queue_new(f->inst.ua, 0, 0, 0, "test send"); assert(f->conn.send_input_q);
+    queue_set_waiter_defer(f->conn.send_input_q, 1);
+    struct ll_entry* e = queue_entry_new(sizeof(struct conn_queue_entry)); assert(e);
+    struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data;
+    ce->peer_node_id = 2; ce->conn = &f->conn;
+    assert(queue_data_put_with_index(f->inst.connections, e) == 0);
+    assert(merkle_sync_init(&f->inst, 0x72, &t_ops, f) == 0);
+}
+
+static void t_clear_queue(struct ll_queue* q) {
+    struct ll_entry* e;
+    while ((e = queue_data_get(q))) { queue_dgram_free(e); queue_entry_free(e); }
+    queue_free(q);
+}
+
+static void t_destroy(struct fixture* f) {
+    merkle_sync_destroy(&f->inst);
+    assert(f->conn.ref_count == 0);
+    t_clear_queue(f->conn.send_input_q); t_clear_queue(f->inst.connections);
+    sqlite3_close(f->inst.topo_sqlite_db);
+    uasync_poll(f->inst.ua, 0); uasync_destroy(f->inst.ua, 0);
+}
+
+static struct ms_session* t_session(struct fixture* f) {
+    struct ms_session* s = ms_create(f->inst.msync, &f->conn, 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;
+    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) {
+    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;
+    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);
+    ms_receive(&f->conn, e);
+}
+
+static void test_malformed_and_old_round(void) {
+    struct fixture f; t_init(&f);
+    struct ms_session* s = t_session(&f);
+    uint8_t short_hashes[4] = {1};
+    t_receive(&f, MS_HASHES, 6, 1, short_hashes, sizeof(short_hashes));
+    assert(!f.callbacks && s->waiting && s->round == 7);
+    t_receive(&f, MS_HASHES, 7, 1, short_hashes, sizeof(short_hashes));
+    assert(f.callbacks == 1 && f.result == MT_ERR_PROTOCOL && s->state == MS_FAILED);
+    t_destroy(&f);
+    puts("PASS: stale round ignored; truncated hashes rejected before reading");
+}
+
+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};
+    t_receive(&f, MS_PAGE, 7, 1, page, sizeof(page));
+    assert(f.callbacks == 1 && f.result == MT_ERR_DATA && s->state == MS_FAILED);
+    t_destroy(&f);
+    puts("PASS: failed apply cannot produce success");
+}
+
+static void test_sibling_walk(void) {
+    struct fixture f; t_init(&f);
+    struct ms_session* s = t_session(&f);
+    uint8_t hashes[32 * 32] = {0}; hashes[0] = 1; hashes[32] = 1;
+    assert(ms_receive_hashes(s, 1, hashes, sizeof(hashes)) == 0);
+    assert(s->depth == 1 && s->stack[1].prefix == 0 && s->stack[0].children == 2);
+    ms_discard_tx(s);
+    memset(hashes, 0, sizeof(hashes));
+    assert(ms_receive_hashes(s, 2, hashes, sizeof(hashes)) == 0);
+    assert(s->depth == 1 && s->stack[1].prefix == (UINT64_C(1) << 59));
+    assert(s->waiting && !f.callbacks);
+    t_destroy(&f);
+    puts("PASS: completing one branch preserves the sibling request");
+}
+
+static void test_cancel_waiters(void) {
+    for (int blocked = 0; blocked < 2; blocked++) {
+        struct fixture f; t_init(&f);
+        struct ms_session* s = t_session(&f);
+        if (blocked) { struct ll_entry* e = queue_entry_new(0); assert(e); assert(queue_data_put(f.conn.send_input_q, e) == 0); }
+        assert(ms_send(s, MS_QUERY, 1, NULL, 0) == 0);
+        assert(s->waiter.internal || s->waiter.call_soon_id);
+        f.conn.links_up = 0;
+        merkle_sync_cancel_peer(&f.inst, 2);
+        assert(!f.inst.msync->sessions && !f.conn.send_input_q->waiter_head);
+        uasync_poll(f.inst.ua, 0);
+        assert(!f.callbacks);
+        t_destroy(&f);
+    }
+    puts("PASS: cancel after links_down removes queued and deferred waiters");
+}
+
+static void test_page_size_and_cursor(void) {
+    struct fixture f; t_init(&f);
+    struct ms_session* s = t_session(&f); 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 && s->tx->len == MT_PAGE_SIZE);
+    ms_discard_tx(s);
+    query[9] = 1; ms_write64(query + 10, 1);
+    assert(ms_serve_query(s, 2, query, sizeof(query)) == MT_ERR_DATA);
+    query[0] = 255;
+    assert(ms_serve_query(s, 3, query, sizeof(query)) == MT_ERR_PROTOCOL);
+    t_destroy(&f);
+    puts("PASS: page bounded; non-advancing cursor and invalid level rejected");
+}
+
+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);
+    uint8_t hash[32], empty[32] = {0};
+    assert(merkle_sync_read_hash(&f.inst, "1001", 0, 0, hash) == 0 && !memcmp(hash, empty, 32));
+    t_destroy(&f);
+    puts("PASS: SQLite failure is reported and cannot publish a root");
+}
+
+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);
+    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");
+    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);
+    assert(!f.inst.msync->sessions && f.callbacks == 1 && f.result == MT_ERR_DISCONNECTED);
+    t_destroy(&f);
+    puts("PASS: 1000 changes coalesce behind backpressure; disconnect completes request with error");
+}
+
+static void test_peer_restart(void) {
+    struct fixture f; t_init(&f);
+    struct ms_session* s = t_session(&f);
+    assert(ms_send(s, MS_QUERY, 1, NULL, 0) == 0);
+    t_receive(&f, MS_WAKE, 0, 0, NULL, 0);
+    assert(s->round > 7 && s->state == MS_BEGIN_WAIT && s->tx && s->tx->dgram[9] == MS_BEGIN);
+    uint8_t hashes[32 * 32] = {0};
+    t_receive(&f, MS_HASHES, 7, 1, hashes, sizeof(hashes));
+    assert(!f.callbacks && s->state == MS_BEGIN_WAIT);
+    t_destroy(&f);
+    puts("PASS: reattaching follower restarts the round; old responses cannot affect it");
+}
+
+static void test_send_rejection(void) {
+    struct fixture f; t_init(&f);
+    struct ms_session* s = t_session(&f);
+    queue_set_size_limit(f.conn.send_input_q, 0);
+    assert(ms_send(s, MS_QUERY, 1, NULL, 0) == 0);
+    uasync_poll(f.inst.ua, 0);
+    assert(f.callbacks == 1 && f.result == MT_ERR_IO && !s->tx);
+    t_destroy(&f);
+    puts("PASS: transport rejection retains ownership; packet is freed exactly once");
+}
+
+int main(void) {
+    debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR);
+    size_t baseline = u_get_allocated_count();
+    test_malformed_and_old_round(); test_rejected_data(); test_sibling_walk(); test_cancel_waiters();
+    test_page_size_and_cursor(); test_readonly_tree(); test_backpressure_and_disconnect(); test_peer_restart();
+    test_send_rejection();
+    assert(u_get_allocated_count() == baseline);
+    puts("ALL PASS: protocol regressions, no tracked allocations leaked");
+    return 0;
+}

tests/test_merkle_sync.c

@@ -35,27 +35,6 @@ static void _ms_ensure_groups(struct UTUN_INSTANCE* inst) {
     }
 }
 
-/* merkle_sync private struct — needed for Stage 2 unit tests.
- * Must match struct ms_session prefix in merkle_sync.c exactly. */
-struct ms_session {
-    struct ms_session* next;
-    char               ns[64];
-    uint64_t           peer;
-    uint8_t            sess_state;
-};
-
-/* merkle_sync private struct — needed for Stage 2 unit tests.
- * Must match struct merkle_sync in merkle_sync.c exactly. */
-struct merkle_sync {
-    struct UTUN_INSTANCE*          inst;
-    uint8_t                       svc_id;
-    const struct merkle_sync_data_ops* ops;
-    void*                         data_ctx;
-    struct ms_session*            sessions;
-    uint8_t                       initialized;
-    void*                         bg_timer;
-};
-
 static int tests_run = 0;
 static int tests_failed = 0;
 static int stage_failures = 0;
@@ -66,7 +45,7 @@ static int stage_failures = 0;
 
 /* ── Stage 2: mock data store ── */
 
-#define MS_MAX_ITEMS 4096
+#define MS_MAX_ITEMS 8192
 
 struct ms_item { uint64_t key; uint32_t val; };
 
@@ -133,80 +112,83 @@ static int _bucket_hash(void* ctx, const char* ns, uint8_t level, uint64_t prefi
     return count;
 }
 
-static int _get_items(void* ctx, const char* ns, uint8_t level, uint64_t prefix,
-                       uint8_t pbytes, uint8_t* buf, size_t* len) {
-    (void)ns; (void)pbytes;
+static int _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) {
+    (void)ns;
     struct ms_data* d = ((struct ms_test_ctx*)ctx)->data;
-    int mask_shift = 64 - (int)level * 5;
-    uint64_t mask = (level == 0) ? 0 : ((mask_shift >= 0 && mask_shift < 64) ? (~0ULL << mask_shift) : UINT64_MAX);
     uint16_t count = 0; size_t off = 2;
+    *next = 0; *more = 0;
     for (int i = 0; i < d->count; i++) {
-        if ((d->items[i].key & mask) != prefix) continue;
-        if (off + 12 > *len) return -2;
+        if (merkle_sync_level_prefix(d->items[i].key, MT_MAX_LEVEL) != prefix
+            || (after_valid && d->items[i].key <= after)) continue;
+        if (off + 12 > *len) { if (!count) return -1; *more = 1; break; }
         memcpy(buf + off, &d->items[i].key, 8); off += 8;
         memcpy(buf + off, &d->items[i].val, 4); off += 4;
-        count++;
+        *next = d->items[i].key; count++;
     }
-    memcpy(buf, &count, 2);
-    *len = off;
-    return 0;
+    memcpy(buf, &count, 2); *len = off; return 0;
 }
 
-static int _apply_items(void* ctx, const char* ns, uint64_t from_peer,
-                         const uint8_t* data, size_t len) {
-    struct ms_test_ctx* tc = (struct ms_test_ctx*)ctx;
+static int _apply_items(void* ctx, const char* ns, uint64_t from_peer, const uint8_t* data, size_t len) {
+    (void)from_peer;
+    struct ms_test_ctx* tc = ctx;
     struct ms_data* d = tc->data;
-    if (len < 2) return 0;
+    if (len < 2) return -1;
     uint16_t count; memcpy(&count, data, 2);
-    const uint8_t* p = data + 2;
+    if (len != 2 + (size_t)count * 12) return -1;
     int changed = 0;
-    uint8_t relay_buf[65536]; uint16_t relay_count = 0; size_t relay_off = 2; /* [count:2][items] */
     for (uint16_t i = 0; i < count; i++) {
-        uint64_t key; memcpy(&key, p, 8); p += 8;
-        uint32_t val; memcpy(&val, p, 4); p += 4;
-        /* check if item is new or different */
+        uint64_t key; uint32_t val;
+        memcpy(&key, data + 2 + (size_t)i * 12, 8);
+        memcpy(&val, data + 10 + (size_t)i * 12, 4);
         int found = 0;
         for (int j = 0; j < d->count; j++) {
-            if (d->items[j].key == key) {
-                if (d->items[j].val != val) { d->items[j].val = val; changed = 1; }
-                found = 1; break;
-            }
-        }
-        if (!found) { _data_insert(d, key, val); changed = 1; }
-        if (changed && relay_off + 12 < sizeof(relay_buf)) {
-            memcpy(relay_buf + relay_off, &key, 8); relay_off += 8;
-            memcpy(relay_buf + relay_off, &val, 4); relay_off += 4;
-            relay_count++;
+            if (d->items[j].key != key) continue;
+            if (d->items[j].val < val) { d->items[j].val = val; changed = 1; }
+            found = 1; break;
         }
+        if (!found) { if (d->count == MS_MAX_ITEMS) return -1; _data_insert(d, key, val); changed = 1; }
     }
-    if (changed && tc->inst) {
+    if (changed) {
         _data_sort(d);
-        for (uint16_t i = 0; i < count; i++) {
-            uint64_t k; memcpy(&k, data + 2 + (size_t)i * 12, 8);
-            merkle_sync_recompute_path(tc->inst, ns, k);
-        }
-        /* relay changed items to other SYNCED sessions */
-        if (relay_count > 0) {
-            memcpy(relay_buf, &relay_count, 2);
-            merkle_sync_broadcast(tc->inst, ns, from_peer, relay_buf, relay_off);
-        }
+        /* Страница всегда относится к одному листу: пересчитываем его один раз. */
+        uint64_t key; memcpy(&key, data + 2, 8);
+        if (merkle_sync_recompute_path(tc->inst, ns, key) < 0) return -1;
     }
     return 0;
 }
 
-static int _apply_update(void* ctx, const char* ns, uint64_t key, uint8_t type,
-                          const uint8_t* data, size_t len) {
-    (void)ctx; (void)ns; (void)key; (void)type; (void)data; (void)len;
-    return 0;
-}
-
 static const struct merkle_sync_data_ops g_test_ops = {
-    .update_bucket_hash = _bucket_hash,
-    .get_items          = _get_items,
-    .apply_items        = _apply_items,
-    .apply_update       = _apply_update,
+    .update_bucket_hash = _bucket_hash, .get_page = _get_page, .apply_items = _apply_items
 };
 
+static const uint8_t* _test_hash(struct UTUN_INSTANCE* inst, const char* ns, uint8_t level, uint64_t prefix) {
+    static uint8_t hash[MT_HASH_SIZE];
+    if (merkle_sync_read_hash(inst, ns, level, prefix, hash) < 0) abort();
+    return hash;
+}
+
+/* Независимое восстановление дерева из набора записей, без чтения индекса. */
+static void _expected_hash(struct ms_test_ctx* tc, const char* ns, uint8_t level, uint64_t prefix, uint8_t hash[MT_HASH_SIZE]) {
+    memset(hash, 0, MT_HASH_SIZE);
+    int present = 0;
+    for (int i = 0; i < tc->data->count; i++)
+        if (merkle_sync_level_prefix(tc->data->items[i].key, level) == prefix) { present = 1; break; }
+    if (!present) return;
+    EVP_MD_CTX* c = EVP_MD_CTX_new();
+    EVP_DigestInit_ex(c, EVP_sha256(), NULL);
+    uint8_t domain[2] = { 0x4d, level }; EVP_DigestUpdate(c, domain, sizeof(domain));
+    if (level == MT_MAX_LEVEL) _bucket_hash(tc, ns, level, prefix, c);
+    else {
+        for (unsigned i = 0; i < MT_BUCKETS; i++) {
+            uint8_t child[MT_HASH_SIZE];
+            _expected_hash(tc, ns, level + 1, prefix | ((uint64_t)i << (64 - (level + 1) * 5)), child);
+            EVP_DigestUpdate(c, child, sizeof(child));
+        }
+    }
+    EVP_DigestFinal_ex(c, hash, NULL); EVP_MD_CTX_free(c);
+}
+
 /* ── Stage 1: prefix arithmetic ── */
 
 static void test_prefix_bytes(void) {
@@ -276,17 +258,7 @@ static void test_level_prefix_bucket_partition(void) {
 
 /* ── Stage 2: tree building tests ── */
 
-static void _ensure_table(sqlite3* db) {
-    sqlite3_exec(db,
-        "CREATE TABLE IF NOT EXISTS merkle_tree_hash ("
-        "  namespace    TEXT NOT NULL,"
-        "  level        INTEGER NOT NULL CHECK(level BETWEEN 1 AND 5),"
-        "  prefix64     INTEGER NOT NULL,"
-        "  hash         BLOB NOT NULL,"
-        "  member_count INTEGER NOT NULL,"
-        "  PRIMARY KEY (namespace, level, prefix64))",
-        NULL, NULL, NULL);
-}
+static void _ensure_table(sqlite3* db) { if (merkle_tree_init(db) < 0) abort(); }
 
 static int _db_count_rows(sqlite3* db, const char* ns) {
     sqlite3_stmt* st = NULL;
@@ -300,9 +272,9 @@ static int _db_count_rows(sqlite3* db, const char* ns) {
 
 static int _db_compare_trees(sqlite3* da, sqlite3* db, const char* ns) {
     sqlite3_stmt* sa = NULL, *sb = NULL;
-    sqlite3_prepare_v2(da, "SELECT level,prefix64,hash,member_count FROM merkle_tree_hash WHERE namespace=? ORDER BY level,prefix64",
+    sqlite3_prepare_v2(da, "SELECT level,prefix64,hash FROM merkle_tree_hash WHERE namespace=? ORDER BY level,prefix64",
                        -1, &sa, NULL);
-    sqlite3_prepare_v2(db, "SELECT level,prefix64,hash,member_count FROM merkle_tree_hash WHERE namespace=? ORDER BY level,prefix64",
+    sqlite3_prepare_v2(db, "SELECT level,prefix64,hash FROM merkle_tree_hash WHERE namespace=? ORDER BY level,prefix64",
                        -1, &sb, NULL);
     sqlite3_bind_text(sa, 1, ns, -1, SQLITE_STATIC);
     sqlite3_bind_text(sb, 1, ns, -1, SQLITE_STATIC);
@@ -316,7 +288,6 @@ static int _db_compare_trees(sqlite3* da, sqlite3* db, const char* ns) {
             (uint64_t)sqlite3_column_int64(sa,1) != (uint64_t)sqlite3_column_int64(sb,1)) {
             mismatch++; break;
         }
-        if (sqlite3_column_int(sa,3) != sqlite3_column_int(sb,3)) { mismatch++; break; }
         if (memcmp(sqlite3_column_blob(sa,2), sqlite3_column_blob(sb,2), MT_HASH_SIZE) != 0) { mismatch++; break; }
     }
     sqlite3_finalize(sa); sqlite3_finalize(sb);
@@ -332,14 +303,13 @@ static void test_tree_empty_ns(void) {
     inst.topo_sqlite_db = db;
     struct ms_data data; _data_init(&data);
     struct ms_test_ctx ctx = { .data = &data, .inst = &inst };
-    struct merkle_sync ms; memset(&ms, 0, sizeof(ms));
-    ms.inst = &inst; ms.ops = &g_test_ops; ms.data_ctx = &ctx; ms.initialized = 1;
-    inst.msync = &ms;
+    inst.ua = uasync_create();
+    if (merkle_sync_init(&inst, 0x72, &g_test_ops, &ctx) != 0) abort();
 
     merkle_sync_recompute_path(&inst, MS_NS_TEST, 0x1234ULL);
     if (_db_count_rows(db, MS_NS_TEST) == 0) PASS(); else FAIL("got %d rows", _db_count_rows(db, MS_NS_TEST));
 
-    sqlite3_close(db);
+    merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db);
 }
 
 static void test_tree_single_item(void) {
@@ -354,26 +324,25 @@ static void test_tree_single_item(void) {
     _data_insert(&data, key, 42);
     _data_sort(&data);
     struct ms_test_ctx ctx = { .data = &data, .inst = &inst };
-    struct merkle_sync ms; memset(&ms, 0, sizeof(ms));
-    ms.inst = &inst; ms.ops = &g_test_ops; ms.data_ctx = &ctx; ms.initialized = 1;
-    inst.msync = &ms;
+    inst.ua = uasync_create();
+    if (merkle_sync_init(&inst, 0x72, &g_test_ops, &ctx) != 0) abort();
 
     merkle_sync_recompute_path(&inst, MS_NS_TEST, key);
     int rows = _db_count_rows(db, MS_NS_TEST);
-    if (rows != 5) { FAIL("expected 5 rows got %d", rows); sqlite3_close(db); return; }
+    if (rows != 6) { FAIL("expected 6 rows got %d", rows); merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db); return; }
 
     /* check hashes are non-zero */
     int ok = 1;
     for (uint8_t lv = 1; lv <= 5 && ok; lv++) {
         uint64_t pf = merkle_sync_level_prefix(key, lv);
-        const uint8_t* h = merkle_sync_get_hash(&inst, MS_NS_TEST, lv, pf);
+        const uint8_t* h = _test_hash(&inst, MS_NS_TEST, lv, pf);
         int zero = 1;
         for (int i = 0; i < MT_HASH_SIZE; i++) if (h[i] != 0) zero = 0;
         if (zero) { FAIL("L%d zero hash", lv); ok = 0; }
     }
     if (ok) PASS();
 
-    sqlite3_close(db);
+    merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db);
 }
 
 static void test_tree_multi_item_same_bucket(void) {
@@ -389,9 +358,8 @@ static void test_tree_multi_item_same_bucket(void) {
     _data_insert(&data, k1, 1); _data_insert(&data, k2, 2);
     _data_sort(&data);
     struct ms_test_ctx ctx = { .data = &data, .inst = &inst };
-    struct merkle_sync ms; memset(&ms, 0, sizeof(ms));
-    ms.inst = &inst; ms.ops = &g_test_ops; ms.data_ctx = &ctx; ms.initialized = 1;
-    inst.msync = &ms;
+    inst.ua = uasync_create();
+    if (merkle_sync_init(&inst, 0x72, &g_test_ops, &ctx) != 0) abort();
 
     /* recompute for both */
     merkle_sync_recompute_path(&inst, MS_NS_TEST, k1);
@@ -399,22 +367,15 @@ static void test_tree_multi_item_same_bucket(void) {
 
     uint64_t pf1 = merkle_sync_level_prefix(k1, 1);
     uint64_t pf2 = merkle_sync_level_prefix(k2, 1);
-    if (pf1 != pf2) { FAIL("L1 prefixes should be equal"); sqlite3_close(db); return; }
+    if (pf1 != pf2) { FAIL("L1 prefixes should be equal"); merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db); return; }
 
     /* verify: both items contribute to same bucket hash */
     uint8_t expected_hash[MT_HASH_SIZE];
-    {
-        EVP_MD_CTX* ctx = EVP_MD_CTX_new();
-        EVP_DigestInit_ex(ctx, EVP_sha256(), NULL);
-        uint8_t ih[MT_HASH_SIZE]; _compute_item_hash(k1, 1, ih); EVP_DigestUpdate(ctx, ih, MT_HASH_SIZE);
-        _compute_item_hash(k2, 2, ih); EVP_DigestUpdate(ctx, ih, MT_HASH_SIZE);
-        EVP_DigestFinal_ex(ctx, expected_hash, NULL);
-        EVP_MD_CTX_free(ctx);
-    }
-    const uint8_t* stored = merkle_sync_get_hash(&inst, MS_NS_TEST, 1, pf1);
+    _expected_hash(&ctx, MS_NS_TEST, 1, pf1, expected_hash);
+    const uint8_t* stored = _test_hash(&inst, MS_NS_TEST, 1, pf1);
     if (memcmp(expected_hash, stored, MT_HASH_SIZE) == 0) PASS(); else FAIL("hash mismatch");
 
-    sqlite3_close(db);
+    merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db);
 }
 
 static void test_tree_different_buckets(void) {
@@ -430,24 +391,23 @@ static void test_tree_different_buckets(void) {
     _data_insert(&data, kA, 100); _data_insert(&data, kB, 200);
     _data_sort(&data);
     struct ms_test_ctx ctx = { .data = &data, .inst = &inst };
-    struct merkle_sync ms; memset(&ms, 0, sizeof(ms));
-    ms.inst = &inst; ms.ops = &g_test_ops; ms.data_ctx = &ctx; ms.initialized = 1;
-    inst.msync = &ms;
+    inst.ua = uasync_create();
+    if (merkle_sync_init(&inst, 0x72, &g_test_ops, &ctx) != 0) abort();
 
     merkle_sync_recompute_path(&inst, "ns", kA);
     merkle_sync_recompute_path(&inst, "ns", kB);
 
     uint64_t pfA = merkle_sync_level_prefix(kA, 1);
     uint64_t pfB = merkle_sync_level_prefix(kB, 1);
-    if (pfA == pfB) { FAIL("same L1 prefix"); sqlite3_close(db); return; }
+    if (pfA == pfB) { FAIL("same L1 prefix"); merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db); return; }
 
-    const uint8_t* hA = merkle_sync_get_hash(&inst, "ns", 1, pfA);
+    const uint8_t* hA = _test_hash(&inst, "ns", 1, pfA);
     uint8_t copyA[MT_HASH_SIZE]; memcpy(copyA, hA, MT_HASH_SIZE);
-    const uint8_t* hB = merkle_sync_get_hash(&inst, "ns", 1, pfB);
+    const uint8_t* hB = _test_hash(&inst, "ns", 1, pfB);
     uint8_t copyB[MT_HASH_SIZE]; memcpy(copyB, hB, MT_HASH_SIZE);
     if (memcmp(copyA, copyB, MT_HASH_SIZE) != 0) PASS(); else FAIL("hashes should differ");
 
-    sqlite3_close(db);
+    merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db);
 }
 
 static void test_tree_delete_empty_bucket(void) {
@@ -461,25 +421,24 @@ static void test_tree_delete_empty_bucket(void) {
     uint64_t k = 0xCCCC000000000000ULL;
     _data_insert(&data, k, 7); _data_sort(&data);
     struct ms_test_ctx ctx = { .data = &data, .inst = &inst };
-    struct merkle_sync ms; memset(&ms, 0, sizeof(ms));
-    ms.inst = &inst; ms.ops = &g_test_ops; ms.data_ctx = &ctx; ms.initialized = 1;
-    inst.msync = &ms;
+    inst.ua = uasync_create();
+    if (merkle_sync_init(&inst, 0x72, &g_test_ops, &ctx) != 0) abort();
 
     merkle_sync_recompute_path(&inst, MS_NS_TEST, k);
-    if (_db_count_rows(db, MS_NS_TEST) != 5) { FAIL("expected 5 rows"); sqlite3_close(db); return; }
+    if (_db_count_rows(db, MS_NS_TEST) != 6) { FAIL("expected 6 rows"); merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db); return; }
 
     /* remove item and recompute */
     data.count = 0;
     merkle_sync_recompute_path(&inst, MS_NS_TEST, k);
-    if (_db_count_rows(db, MS_NS_TEST) != 0) { FAIL("expected 0 rows got %d", _db_count_rows(db, MS_NS_TEST)); sqlite3_close(db); return; }
+    if (_db_count_rows(db, MS_NS_TEST) != 0) { FAIL("expected 0 rows got %d", _db_count_rows(db, MS_NS_TEST)); merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db); return; }
 
     uint64_t pf = merkle_sync_level_prefix(k, 3);
-    const uint8_t* h = merkle_sync_get_hash(&inst, MS_NS_TEST, 3, pf);
+    const uint8_t* h = _test_hash(&inst, MS_NS_TEST, 3, pf);
     int zero = 1;
     for (int i = 0; i < MT_HASH_SIZE; i++) if (h[i] != 0) zero = 0;
     if (zero) PASS(); else FAIL("hash not zero after delete");
 
-    sqlite3_close(db);
+    merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db);
 }
 
 static void test_tree_consistency(void) {
@@ -500,9 +459,8 @@ static void test_tree_consistency(void) {
     _data_sort(&data);
 
     struct ms_test_ctx ctx = { .data = &data, .inst = &inst };
-    struct merkle_sync ms; memset(&ms, 0, sizeof(ms));
-    ms.inst = &inst; ms.ops = &g_test_ops; ms.data_ctx = &ctx; ms.initialized = 1;
-    inst.msync = &ms;
+    inst.ua = uasync_create();
+    if (merkle_sync_init(&inst, 0x72, &g_test_ops, &ctx) != 0) abort();
 
     for (int i = 0; i < data.count; i++)
         merkle_sync_recompute_path(&inst, "ns", data.items[i].key);
@@ -519,14 +477,10 @@ static void test_tree_consistency(void) {
         uint64_t pf = (uint64_t)sqlite3_column_int64(st, 1);
 
         /* compute expected hash from data */
-        EVP_MD_CTX* c = EVP_MD_CTX_new();
-        EVP_DigestInit_ex(c, EVP_sha256(), NULL);
-        _bucket_hash(&ctx, "ns", lv, pf, c);
         uint8_t expected[MT_HASH_SIZE];
-        EVP_DigestFinal_ex(c, expected, NULL);
-        EVP_MD_CTX_free(c);
+        _expected_hash(&ctx, "ns", lv, pf, expected);
 
-        const uint8_t* stored = merkle_sync_get_hash(&inst, "ns", lv, pf);
+        const uint8_t* stored = _test_hash(&inst, "ns", lv, pf);
         if (memcmp(expected, stored, MT_HASH_SIZE) != 0) {
             FAIL("mismatch L%d P%016llx", lv, (unsigned long long)pf);
             ok = 0;
@@ -535,7 +489,7 @@ static void test_tree_consistency(void) {
     sqlite3_finalize(st);
     if (ok) PASS();
 
-    sqlite3_close(db);
+    merkle_sync_destroy(&inst); uasync_destroy(inst.ua, 0); sqlite3_close(db);
 }
 
 static int run_stage1(void) {
@@ -584,6 +538,7 @@ static int _cond_done(void) { return done_sync; }
 static void _on_sync_done(uint64_t peer, const char* ns, int result, void* arg) {
     (void)arg; done_sync = 1;
     printf("  sync_done: peer=%016llx ns=%s result=%d\n", (unsigned long long)peer, ns, result);
+    if (result != MT_OK) { tests_failed++; stage_failures++; }
 }
 
 static int _write_cfg(const char* path, const char* fmt, ...) {
@@ -860,12 +815,9 @@ static void test_randomized_two(void) {
         while (sqlite3_step(st) == SQLITE_ROW && ok) {
             uint8_t lv = (uint8_t)sqlite3_column_int(st, 0);
             uint64_t pf = (uint64_t)sqlite3_column_int64(st, 1);
-            EVP_MD_CTX* c = EVP_MD_CTX_new();
-            EVP_DigestInit_ex(c, EVP_sha256(), NULL);
-            _bucket_hash(&ctx_a, MS_NS_RND, lv, pf, c);
             uint8_t expected[MT_HASH_SIZE];
-            EVP_DigestFinal_ex(c, expected, NULL); EVP_MD_CTX_free(c);
-            const uint8_t* stored = merkle_sync_get_hash(i_a, MS_NS_RND, lv, pf);
+            _expected_hash(&ctx_a, MS_NS_RND, lv, pf, expected);
+            const uint8_t* stored = _test_hash(i_a, MS_NS_RND, lv, pf);
             uint8_t scopy[MT_HASH_SIZE]; memcpy(scopy, stored, MT_HASH_SIZE);
             if (memcmp(expected, scopy, MT_HASH_SIZE) != 0) { ok = 0; }
         }
@@ -935,8 +887,10 @@ static void test_randomized_three(void) {
         /* C→A: C syncs with A, gets A∪B, A broadcasts C's new items to B (relay) */
         merkle_sync_start(i_c, i_a->node_id, MS_NS_RND, _on_sync_done, NULL);
         if (!_wait_for("sync C", _cond_done, PHASE_TIMEOUT_TB)) { FAIL("timeout C iter %d", iter); _intg_cleanup(); break; }
-        /* wait for broadcast relay */
-        for (int k = 0; k < 100; k++) uasync_poll(i_ua, 10);
+        /* Автоматический следующий раунд A↔B завершается по условию, не по числу poll(). */
+        if (!_wait_for("automatic convergence of all peers", _cond_all_synced, PHASE3_TIMEOUT_TB)) {
+            FAIL("automatic sync timeout iter %d", iter); _intg_cleanup(); break;
+        }
         if (_intg_compare_data(&data_a, &data_b) != 0 || _intg_compare_data(&data_b, &data_c) != 0)
             { FAIL("data mismatch iter %d", iter); _intg_cleanup(); break; }
         if (_db_compare_trees(i_a->topo_sqlite_db, i_b->topo_sqlite_db, MS_NS_RND) != 0 ||
@@ -955,7 +909,6 @@ static void test_cancel(void) {
     _intg_ins_many(&data_b, i_b, 0, 5);
     done_sync = 0;
     merkle_sync_start(i_a, i_b->node_id, MS_NS_TEST, _on_sync_done, NULL);
-    uasync_poll(i_ua, 1);
     merkle_sync_cancel(i_a, i_b->node_id, MS_NS_TEST);
     for (int i = 0; i < 100 && i_phase == 0; i++) uasync_poll(i_ua, 10);
     if (!done_sync) PASS(); else FAIL("done_cb was called");
@@ -967,14 +920,14 @@ static void _sw_done1(uint64_t p, const char* n, int r, void* a) { (void)p;(void
 static void _sw_done2(uint64_t p, const char* n, int r, void* a) { (void)p;(void)n;(void)r;(void)a; sw_cb2_called = 1; }
 
 static void test_start_overwrite(void) {
-    TEST("start overwrite — cb1 replaced by cb2, only cb2 fires");
+    TEST("concurrent starts — both callbacks complete");
     if (_intg_init_two() != 0) { FAIL("setup failed"); return; }
     _intg_ins_many(&data_b, i_b, 0, 3);
     sw_cb1_called = sw_cb2_called = 0;
     merkle_sync_start(i_a, i_b->node_id, MS_NS_TEST, _sw_done1, NULL);
     merkle_sync_start(i_a, i_b->node_id, MS_NS_TEST, _sw_done2, NULL);
     for (int i = 0; i < 500 && !sw_cb2_called && i_phase == 0; i++) uasync_poll(i_ua, 10);
-    if (sw_cb2_called) PASS(); else FAIL("cb1=%d cb2=%d", sw_cb1_called, sw_cb2_called);
+    if (sw_cb1_called && sw_cb2_called) PASS(); else FAIL("cb1=%d cb2=%d", sw_cb1_called, sw_cb2_called);
     _intg_cleanup();
 }
 
@@ -1277,10 +1230,10 @@ static void _str_verify(void) {
     }
 
     /* root (level=1, prefix=0) hashes identical */
-    const uint8_t* root0 = merkle_sync_get_hash(s->inst[STRESS_HUB], MS_NS_STRESS, 1, 0);
+    const uint8_t* root0 = _test_hash(s->inst[STRESS_HUB], MS_NS_STRESS, 1, 0);
     uint8_t root_copy[MT_HASH_SIZE]; memcpy(root_copy, root0, MT_HASH_SIZE);
     for (int i = 1; i < STRESS_N && ok; i++) {
-        const uint8_t* ri = merkle_sync_get_hash(s->inst[i], MS_NS_STRESS, 1, 0);
+        const uint8_t* ri = _test_hash(s->inst[i], MS_NS_STRESS, 1, 0);
         if (memcmp(root_copy, ri, MT_HASH_SIZE) != 0) { printf("  ROOT HASH MISMATCH inst[%d]\n", i); ok = 0; }
     }
 
@@ -1294,12 +1247,9 @@ static void _str_verify(void) {
         while (sqlite3_step(st) == SQLITE_ROW && ok) {
             uint8_t lv = (uint8_t)sqlite3_column_int(st, 0);
             uint64_t pf = (uint64_t)sqlite3_column_int64(st, 1);
-            EVP_MD_CTX* c = EVP_MD_CTX_new();
-            EVP_DigestInit_ex(c, EVP_sha256(), NULL);
-            _bucket_hash(&s->ctx[i], MS_NS_STRESS, lv, pf, c);
             uint8_t expected[MT_HASH_SIZE];
-            EVP_DigestFinal_ex(c, expected, NULL); EVP_MD_CTX_free(c);
-            const uint8_t* stored = merkle_sync_get_hash(s->inst[i], MS_NS_STRESS, lv, pf);
+            _expected_hash(&s->ctx[i], MS_NS_STRESS, lv, pf, expected);
+            const uint8_t* stored = _test_hash(s->inst[i], MS_NS_STRESS, lv, pf);
             uint8_t scopy[MT_HASH_SIZE]; memcpy(scopy, stored, MT_HASH_SIZE);
             if (memcmp(expected, scopy, MT_HASH_SIZE) != 0) {
                 printf("  consistency fail inst[%d] L%d P%016llx\n", i, lv, (unsigned long long)pf);
@@ -1320,12 +1270,12 @@ static void _str_verify(void) {
 static void _str_check_roots(void) {
     struct str_state* s = gs;
     if (!s) return;
-    const uint8_t* root0 = merkle_sync_get_hash(s->inst[STRESS_HUB], MS_NS_STRESS, 1, 0);
+    const uint8_t* root0 = _test_hash(s->inst[STRESS_HUB], MS_NS_STRESS, 1, 0);
     uint8_t root_copy[MT_HASH_SIZE];
     memcpy(root_copy, root0, MT_HASH_SIZE);
     int converged = 1;
     for (int i = 1; i < STRESS_N; i++) {
-        const uint8_t* ri = merkle_sync_get_hash(s->inst[i], MS_NS_STRESS, 1, 0);
+        const uint8_t* ri = _test_hash(s->inst[i], MS_NS_STRESS, 1, 0);
         if (memcmp(root_copy, ri, MT_HASH_SIZE) != 0) { converged = 0; break; }
     }
     if (converged) {
@@ -1387,11 +1337,38 @@ static void test_stress_spam(void) {
     PASS();
     _str_cleanup();
 }
+static int _cond_pair_equal(void) {
+    if (data_a.count != data_b.count) return 0;
+    for (int i = 0; i < data_a.count; i++)
+        if (data_a.items[i].key != data_b.items[i].key || data_a.items[i].val != data_b.items[i].val) return 0;
+    return _db_compare_trees(i_a->topo_sqlite_db, i_b->topo_sqlite_db, MS_NS_TEST) == 0;
+}
+
+static void test_large_leaf_and_automatic_update(void) {
+    TEST("6000 records in one leaf (>64K), then automatic update without start/broadcast");
+    if (_intg_init_two() != 0) { FAIL("setup failed"); return; }
+    for (uint64_t i = 0; i < 6000; i++) _data_insert(&data_b, i, (uint32_t)i);
+    _data_sort(&data_b);
+    if (merkle_sync_recompute_path(i_b, MS_NS_TEST, 0) < 0) { FAIL("recompute failed"); _intg_cleanup(); return; }
+    done_sync = 0;
+    merkle_sync_start(i_a, i_b->node_id, MS_NS_TEST, _on_sync_done, NULL);
+    if (!_wait_for("large leaf", _cond_done, PHASE_TIMEOUT_TB) || !_cond_pair_equal()) {
+        FAIL("large leaf did not converge"); _intg_cleanup(); return;
+    }
+    data_b.items[0].val = 9999;
+    merkle_sync_recompute_path(i_b, MS_NS_TEST, 0);
+    if (!_wait_for("automatic change", _cond_pair_equal, PHASE_TIMEOUT_TB)) {
+        FAIL("automatic change did not propagate"); _intg_cleanup(); return;
+    }
+    PASS(); _intg_cleanup();
+}
+
 static int run_stage6(void) {
     printf("--- Stage 6: protocol edge cases ---\n");
     stage_failures = 0;
     test_cancel();
     test_start_overwrite();
+    test_large_leaf_and_automatic_update();
     return stage_failures;
 }
 
@@ -1406,6 +1383,7 @@ int main(void) {
     printf("=== test_merkle_sync ===\n");
     debug_config_init();
     debug_set_level(DEBUG_LEVEL_ERROR);
+    if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG);
 
     if (run_stage1() != 0) { printf("=== Stage 1 FAILED ===\n"); return 1; }
     printf("=== Stage 1 PASSED (%d tests) ===\n\n", tests_run);