utun2/tests/test_merkle_protocol.c

/* Проверяем границы автомата на точных 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 && s->state == MS_FAILING);
    uasync_poll(f.inst.ua, 0);
    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 && s->state == MS_FAILING);
    uasync_poll(f.inst.ua, 0);
    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");
}

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);
}

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};
    t_receive(&f, MS_PAGE, 7, 1, page, sizeof(page));
    assert(!f.callbacks && s->state == MS_FAILING);
    uasync_poll(f.inst.ua, 0);
    assert(f.callbacks == 1 && f.result == MT_ERR_DATA && !f.inst.msync->sessions);
    assert(f.conn.send_input_q->head && f.conn.send_input_q->head->dgram[9] == MS_ERROR);
    t_destroy(&f);
    puts("PASS: ERROR reaches transport before cancellation inside callback");
}

static void test_blocked_error_timeout(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);
    struct ms_session* s = t_session(&f); s->requests->cb = t_cancel_done;
    ms_fail(s, MT_ERR_DATA, "injected apply failure", 1);
    assert(s->deadline && s->state == MS_FAILING && !f.callbacks);
    ms_stop_deadline(s); ms_timeout(s);
    assert(f.callbacks == 1 && f.result == MT_ERR_TIMEOUT && !f.inst.msync->sessions);
    uasync_poll(f.inst.ua, 0);
    assert(f.callbacks == 1 && !f.conn.send_input_q->waiter_head);
    t_destroy(&f);
    puts("PASS: blocked terminal packet times out once and releases waiter/session");
}

static int pull_created, pull_committed, pull_released;
static int t_begin_pull(void* ctx, const char* ns, uint64_t peer, uint64_t round, void** pull) {
    (void)ctx; (void)ns; (void)peer; (void)round;
    *pull = u_malloc(1); assert(*pull); pull_created++; return 0;
}
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) {
    (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;
    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;
    assert(ms_walk_next(s) == 0 && !s->pull && pull_committed == 1 && pull_released == 1);
    ms_discard_tx(s);
    assert(ms_pull_start(s) == 0);
    ms_discard_tx(s); assert(ms_begin(s) == 0 && !s->pull && pull_released == 2);
    ms_discard_tx(s); assert(ms_pull_start(s) == 0);
    ms_conn_status(&f.conn, ETCP_CONN_STATUS_DOWN, f.inst.msync);
    assert(pull_created == 3 && pull_committed == 1 && pull_released == 3);
    t_destroy(&f);
    puts("PASS: pull commits once; restart and disconnect abort exactly their own context");
}

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();
    test_error_handoff(); test_blocked_error_timeout(); test_pull_lifecycle();
    assert(u_get_allocated_count() == baseline);
    puts("ALL PASS: protocol regressions, no tracked allocations leaked");
    return 0;
}