From 3aa0cd4c0635eced88655dac257756a9bf8d22c0 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 7 Oct 2026 23:50:02 +0200 Subject: [PATCH] tests: add standalone Merkle sync reconnect and triangle stress test --- tests/Makefile.am | 10 + tests/test_merkle_sync_stress.c | 515 ++++++++++++++++++++++++++++++++ 2 files changed, 525 insertions(+) create mode 100644 tests/test_merkle_sync_stress.c diff --git a/tests/Makefile.am b/tests/Makefile.am index b90ef171..093b5940 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -145,6 +145,16 @@ check_PROGRAMS += test_speex_aec # Долгие тесты: запускаются только вручную noinst_PROGRAMS = +EXTRA_PROGRAMS = test_merkle_sync_stress +CLEANFILES = test_merkle_sync_stress$(EXEEXT) + +# Разрывы двух узлов и race трёх SQLite-баз; не входит в all/check/check.sh. +.PHONY: check-merkle-sync-stress +check-merkle-sync-stress: test_merkle_sync_stress$(EXEEXT) + ./test_merkle_sync_stress$(EXEEXT) + +test_merkle_sync_stress_SOURCES = test_merkle_sync_stress.c +test_merkle_sync_stress_LDADD = $(LIBUTUN) $(COMMON_LIBS) # Basic includes AM_CFLAGS = -g diff --git a/tests/test_merkle_sync_stress.c b/tests/test_merkle_sync_stress.c new file mode 100644 index 00000000..aba221df --- /dev/null +++ b/tests/test_merkle_sync_stress.c @@ -0,0 +1,515 @@ +/* Долгий сетевой тест Merkle sync, запускается только вручную. + * Два узла: force-close -> 1..100 изменений -> reconnect -> полная сверка SQLite. + * Три узла: треугольник, записи во время обмена, периодические проверки сходимости. + * Модель: key BE64, version BE64, value[32]; побеждает большая версия. + * Эталон меняется только генератором локальных записей, никогда при сетевом приёме. */ +#include +#include +#include +#include + +#include "../lib/debug_config.h" +#include "../lib/mem.h" +#include "../lib/platform_compat.h" +#include "../lib/socket_compat.h" +#include "../src/utun_instance.h" +#include "../src/chat/merkle_sync.h" +#include "../src/routing_layer/topo_node.h" +#include "../src/transport_layer/etcp.h" +#include "../src/transport_layer/etcp_connections.h" +#include "../src/transport_layer/node_conn_direct.h" + +#define RECORDS 10000 +#define MAX_BATCH 100 +#define RACE_BATCHES 1000 +#define RACE_BURST 20 +#define WIRE_SIZE 48 +#define TEST_NS "1001" +#define TEST_SVC 0x72 +#define SEED UINT64_C(0x4d65726b6c653130) +#define SECOND_TB UINT64_C(10000) + +struct record { uint8_t wire[WIRE_SIZE]; }; + +struct node { + struct UTUN_INSTANCE* inst; + sqlite3_stmt* scan; + sqlite3_stmt* write; + unsigned port; + uint64_t pages, continued, received, applied, stale; +}; + +static struct { + struct UASYNC* ua; + struct node nodes[3]; + struct NODE_CONN_DIRECT* links[3][3]; + struct record expected[RECORDS]; + uint64_t rng, version, started, disconnect_max; + int count, inserted, phase, batch, reconnects, callbacks; + int race_left, race_writes, race_overlap, page_interleaves; + uint64_t last_pages; + void* producer; +} test; + +/* На ошибке сохраняем контекст, корни, объёмы и статистику каждого узла. */ +static void fail(const char* expression, int line) { + DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stress: FAIL seed=%016" PRIx64 " phase=%d batch=%d inserted=%d version=%" PRIu64 + " line=%d: %s", SEED, test.phase, test.batch, test.inserted, test.version, line, expression); + for (int i = 0; i < test.count; i++) { + struct node* n = &test.nodes[i]; + if (!n->inst) continue; + DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stress: node=%d pages=%" PRIu64 " continued=%" PRIu64 " received=%" PRIu64 + " applied=%" PRIu64 " stale=%" PRIu64 " SQL=%s", i, n->pages, n->continued, n->received, n->applied, + n->stale, sqlite3_errmsg(n->inst->topo_sqlite_db)); + uint8_t root[MT_HASH_SIZE]; + if (merkle_sync_read_hash(n->inst, MT_DATA_MEMBERS, TEST_NS, 0, 0, root) == 0) + log_dump(DEBUG_LEVEL_ERROR, DEBUG_CATEGORY_DEBUG, "root", root, sizeof(root)); + } + exit(EXIT_FAILURE); +} + +#define CHECK(expr) do { if (!(expr)) fail(#expr, __LINE__); } while (0) + +static uint64_t read64(const uint8_t* p) { + uint64_t value = 0; + for (int i = 0; i < 8; i++) value = (value << 8) | p[i]; + return value; +} + +static void write64(uint8_t* p, uint64_t value) { + for (int i = 0; i < 8; i++) p[i] = (uint8_t)(value >> (56 - 8 * i)); +} + +/* SplitMix64 не зависит от платформенной реализации rand(). */ +static uint64_t next_random(void) { + uint64_t z = (test.rng += UINT64_C(0x9e3779b97f4a7c15)); + z = (z ^ (z >> 30)) * UINT64_C(0xbf58476d1ce4e5b9); + z = (z ^ (z >> 27)) * UINT64_C(0x94d049bb133111eb); + return z ^ (z >> 31); +} + +static int compare_records(const void* a, const void* b) { return memcmp(a, b, 8); } + +static void shuffled_order(int order[RECORDS]) { + for (int i = 0; i < RECORDS; i++) order[i] = i; + for (int i = RECORDS - 1; i > 0; i--) { + int j = (int)(next_random() % (uint64_t)(i + 1)); + int tmp = order[i]; + order[i] = order[j]; + order[j] = tmp; + } +} + +/* BLOB-ключи дают unsigned-порядок SQLite, включая старший бит node_id. */ +static void scan_leaf(struct node* n, uint64_t prefix, int after_valid, const merkle_key_t* after) { + uint8_t first[8], last[8], zero[8] = {0}; + write64(first, prefix); + write64(last, prefix | ((UINT64_C(1) << (64 - MT_MAX_LEVEL * 5)) - 1)); + CHECK(sqlite3_bind_blob(n->scan, 1, first, 8, SQLITE_TRANSIENT) == SQLITE_OK); + CHECK(sqlite3_bind_blob(n->scan, 2, last, 8, SQLITE_TRANSIENT) == SQLITE_OK); + CHECK(sqlite3_bind_int(n->scan, 3, after_valid) == SQLITE_OK); + CHECK(sqlite3_bind_blob(n->scan, 4, after_valid ? after->bytes : zero, 8, SQLITE_TRANSIENT) == SQLITE_OK); +} + +/* Каноническое представление записи без padding и платформенного byte order. */ +static void read_record(sqlite3_stmt* st, uint8_t wire[WIRE_SIZE]) { + CHECK(sqlite3_column_bytes(st, 0) == 8 && sqlite3_column_bytes(st, 2) == 32); + CHECK(sqlite3_column_int64(st, 1) > 0); + memcpy(wire, sqlite3_column_blob(st, 0), 8); + write64(wire + 8, (uint64_t)sqlite3_column_int64(st, 1)); + memcpy(wire + 16, sqlite3_column_blob(st, 2), 32); +} + +/* Хеш модели вычисляется исключительно из локальной базы узла. */ +static int leaf_hash(void* ctx, const char* ns, uint8_t level, uint64_t prefix, EVP_MD_CTX* sha) { + struct node* n = ctx; + CHECK(!strcmp(ns, TEST_NS) && level == MT_MAX_LEVEL); + scan_leaf(n, prefix, 0, NULL); + int rc, count = 0; + while ((rc = sqlite3_step(n->scan)) == SQLITE_ROW) { + uint8_t wire[WIRE_SIZE]; + read_record(n->scan, wire); + CHECK(EVP_DigestUpdate(sha, wire, sizeof(wire)) == 1); + count++; + } + CHECK(rc == SQLITE_DONE); + CHECK(sqlite3_reset(n->scan) == SQLITE_OK); + return count; +} + +/* Страница ограничена ёмкостью движка; дополнительная строка определяет more. */ +static int get_page(void* ctx, const char* ns, uint64_t prefix, int after_valid, const merkle_key_t* after, + uint8_t* data, size_t* len, merkle_key_t* next, int* more) { + struct node* n = ctx; + CHECK(!strcmp(ns, TEST_NS) && *len >= WIRE_SIZE); + scan_leaf(n, prefix, after_valid, after); + size_t used = 0; + int rc; + memset(next, 0, sizeof(*next)); + while ((rc = sqlite3_step(n->scan)) == SQLITE_ROW && used + WIRE_SIZE <= *len) { + read_record(n->scan, data + used); + memcpy(next->bytes, data + used, 8); + used += WIRE_SIZE; + } + CHECK(rc == SQLITE_ROW || rc == SQLITE_DONE); + *more = rc == SQLITE_ROW; + *len = used; + CHECK(sqlite3_reset(n->scan) == SQLITE_OK); + n->pages++; + n->continued += *more; + return 0; +} + +/* Старая или повторно доставленная версия не должна перезаписывать новую. */ +static int put_record(struct node* n, const uint8_t wire[WIRE_SIZE]) { + CHECK(read64(wire + 8) > 0 && read64(wire + 8) <= INT64_MAX); + CHECK(sqlite3_bind_blob(n->write, 1, wire, 8, SQLITE_STATIC) == SQLITE_OK); + CHECK(sqlite3_bind_int64(n->write, 2, (sqlite3_int64)read64(wire + 8)) == SQLITE_OK); + CHECK(sqlite3_bind_blob(n->write, 3, wire + 16, 32, SQLITE_STATIC) == SQLITE_OK); + CHECK(sqlite3_step(n->write) == SQLITE_DONE); + int changed = sqlite3_changes(n->inst->topo_sqlite_db); + CHECK(sqlite3_reset(n->write) == SQLITE_OK); + return changed; +} + +/* Принятые записи и индекс публикуются одной транзакцией; changed будит другие сессии. */ +static int apply_page(void* ctx, const char* ns, uint64_t peer, const uint8_t* data, size_t len) { + struct node* n = ctx; + CHECK(!strcmp(ns, TEST_NS) && peer != n->inst->node_id && len % WIRE_SIZE == 0); + if (!len) return 0; + uint64_t prefix = merkle_sync_level_prefix(read64(data), MT_MAX_LEVEL); + for (size_t off = 0; off < len; off += WIRE_SIZE) { + CHECK(merkle_sync_level_prefix(read64(data + off), MT_MAX_LEVEL) == prefix); + CHECK(off == 0 || memcmp(data + off - WIRE_SIZE, data + off, 8) < 0); + CHECK(read64(data + off + 8) > 0 && read64(data + off + 8) <= INT64_MAX); + } + sqlite3* db = n->inst->topo_sqlite_db; + CHECK(sqlite3_exec(db, "BEGIN", NULL, NULL, NULL) == SQLITE_OK); + int changed = 0; + for (size_t off = 0; off < len; off += WIRE_SIZE) changed += put_record(n, data + off); + if (changed) CHECK(merkle_sync_recompute_path(n->inst, MT_DATA_MEMBERS, ns, prefix) == 1); + CHECK(sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) == SQLITE_OK); + if (changed) merkle_sync_changed(n->inst, MT_DATA_MEMBERS, ns); + n->received += len / WIRE_SIZE; + n->applied += changed; + n->stale += len / WIRE_SIZE - changed; + return 0; +} + +static const struct merkle_sync_data_ops ops = { + .update_bucket_hash = leaf_hash, .get_page = get_page, .apply_items = apply_page +}; + +/* ОС выбирает порт; ядро создаётся сразу после освобождения пробного сокета. */ +static unsigned free_port(void) { + socket_t fd = socket(AF_INET, SOCK_DGRAM, 0); + CHECK(fd != SOCKET_INVALID); + struct sockaddr_in addr = {0}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + CHECK(bind(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0); + socklen_t len = sizeof(addr); + CHECK(getsockname(fd, (struct sockaddr*)&addr, &len) == 0); + unsigned port = ntohs(addr.sin_port); + socket_close_wrapper(fd); + return port; +} + +/* Отдельное ядро и SQLite :memory:, без чата, TUN и фонового автоподключения групп. */ +static void create_node(int index) { + struct node* n = &test.nodes[index]; + struct SC_MYKEYS keys; + /* Фиксированные тестовые ключи сохраняют роли leader/follower между прогонами. */ + for (int i = 0; i < 32; i++) keys.private_key[i] = (uint8_t)(49 + index * 47 + i * 13); + CHECK(sc_compute_public_key_from_private(keys.private_key, keys.public_key) == SC_OK); + char private_key[65], public_key[65], config[1024]; + for (int i = 0; i < 32; i++) { + snprintf(private_key + 2 * i, 3, "%02x", keys.private_key[i]); + snprintf(public_key + 2 * i, 3, "%02x", keys.public_key[i]); + } + n->port = free_port(); + snprintf(config, sizeof(config), "[global]\nmy_private_key=%s\nmy_public_key=%s\nmy_node_id=0x%016" PRIx64 + "\nkeepalive_adaptive=0\n[server: udp]\naddr=127.0.0.1:%u\ntype=public\n[allowed_keys]\nallow_all=1\n", + private_key, public_key, sc_derive_node_id_from_pubkey(keys.public_key), n->port); + n->inst = utun_instance_create_from_str(test.ua, config); + CHECK(n->inst && utun_core_start(n->inst) == 0); + sqlite3* db = n->inst->topo_sqlite_db; + CHECK(sqlite3_exec(db, "CREATE TABLE stress_records(key BLOB PRIMARY KEY CHECK(length(key)=8)," + "version INTEGER NOT NULL,value BLOB NOT NULL CHECK(length(value)=32)) WITHOUT ROWID", NULL, NULL, NULL) == SQLITE_OK); + CHECK(sqlite3_prepare_v2(db, "SELECT key,version,value FROM stress_records WHERE key>=?1 AND key<=?2 " + "AND (?3=0 OR key>?4) ORDER BY key", -1, &n->scan, NULL) == SQLITE_OK); + CHECK(sqlite3_prepare_v2(db, "INSERT INTO stress_records VALUES(?1,?2,?3) ON CONFLICT(key) DO UPDATE SET " + "version=excluded.version,value=excluded.value WHERE excluded.version>stress_records.version", + -1, &n->write, NULL) == SQLITE_OK); + CHECK(merkle_sync_init(n->inst, TEST_SVC, &ops, n) == 0); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: node=%d ready id=%016" PRIx64 " port=%u", index, n->inst->node_id, n->port); +} + +/* Обе стороны держат ровно по одному handle: force_close не оставляет чужих владельцев. */ +static void open_link(int a, int b) { + CHECK(!test.links[a][b]); + struct node* dest = &test.nodes[b]; + struct TOPO_ADDR4 address = { .addr = {127, 0, 0, 1}, .port = dest->port, .protocol = TOPO_PROTO_UDP }; + struct TOPO_NODE node = { .node_id = dest->inst->node_id, .v4_addrs = &address }; + memcpy(node.public_key, dest->inst->my_keys.public_key, 32); + CHECK(node_conn_direct_open_node(test.nodes[a].inst, node.node_id, NULL, NULL, &test.links[a][b], &node, NULL) != NCD_ERR); +} + +static int transport_ready(int a, int b) { + struct ETCP_CONN* c = instance_find_conn(test.nodes[a].inst, test.nodes[b].inst->node_id); + return c && c->initialized && c->links_up && !c->close_requested && c->state != 2; +} + +/* Инициируем одну сторону, а входящий транспорт берём во владение после UP. */ +static void connect_pair(int a, int b) { + open_link(a, b); + uint64_t deadline = get_time_tb() + 10 * SECOND_TB; + while (!transport_ready(a, b) || !transport_ready(b, a)) { + CHECK(get_time_tb() < deadline); + uasync_poll(test.ua, 1); + } + open_link(b, a); +} + +/* Только обработка отложенного close, без ожидания keepalive или fin_wait. */ +static void disconnect_pair(int a, int b) { + uint64_t started = get_time_tb(); + CHECK(test.links[a][b] && test.links[b][a]); + node_conn_direct_force_close(test.links[a][b]); + node_conn_direct_force_close(test.links[b][a]); + test.links[a][b] = test.links[b][a] = NULL; + while (instance_find_conn(test.nodes[a].inst, test.nodes[b].inst->node_id) + || instance_find_conn(test.nodes[b].inst, test.nodes[a].inst->node_id)) { + CHECK(get_time_tb() - started < SECOND_TB); + uasync_poll(test.ua, 0); + } + uint64_t elapsed = get_time_tb() - started; + if (elapsed > test.disconnect_max) test.disconnect_max = elapsed; + test.reconnects++; +} + +static void sync_done(uint64_t peer, const char* ns, int result, void* arg) { + (void)arg; + if (result != MT_OK) DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stress: sync peer=%016" PRIx64 " ns=%s result=%d", peer, ns, result); + CHECK(result == MT_OK && !strcmp(ns, TEST_NS)); + test.callbacks++; +} + +/* Запрос контрольной точки только у leader; follower создаёт сессию по входящему BEGIN. */ +static void start_pair(int a, int b) { + if (test.nodes[a].inst->node_id > test.nodes[b].inst->node_id) { int tmp = a; a = b; b = tmp; } + CHECK(merkle_sync_start(test.nodes[a].inst, MT_DATA_MEMBERS, test.nodes[b].inst->node_id, TEST_NS, sync_done, NULL) == 0); +} + +static int roots_equal(void) { + uint8_t first[MT_HASH_SIZE], current[MT_HASH_SIZE]; + CHECK(merkle_sync_read_hash(test.nodes[0].inst, MT_DATA_MEMBERS, TEST_NS, 0, 0, first) == 0); + for (int i = 1; i < test.count; i++) { + CHECK(merkle_sync_read_hash(test.nodes[i].inst, MT_DATA_MEMBERS, TEST_NS, 0, 0, current) == 0); + if (memcmp(first, current, sizeof(first))) return 0; + } + return 1; +} + +/* Сравнение с независимым эталоном обнаружит одинаково неверные базы и лишние записи. */ +static void verify_databases(void) { + for (int i = 0; i < test.count; i++) { + sqlite3_stmt* st = NULL; + CHECK(sqlite3_prepare_v2(test.nodes[i].inst->topo_sqlite_db, + "SELECT key,version,value FROM stress_records ORDER BY key", -1, &st, NULL) == SQLITE_OK); + int count = 0; + for (int r = 0; r < RECORDS; r++) { + const uint8_t* expected = test.expected[r].wire; + if (!read64(expected + 8)) continue; + CHECK(sqlite3_step(st) == SQLITE_ROW); + uint8_t actual[WIRE_SIZE]; + read_record(st, actual); + if (memcmp(actual, expected, WIRE_SIZE)) { + DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "stress: data mismatch node=%d record=%d key=%016" PRIx64 + " expected_version=%" PRIu64 " actual_key=%016" PRIx64 " actual_version=%" PRIu64, + i, r, read64(expected), read64(expected + 8), read64(actual), read64(actual + 8)); + log_dump(DEBUG_LEVEL_ERROR, DEBUG_CATEGORY_DEBUG, "expected", expected, WIRE_SIZE); + log_dump(DEBUG_LEVEL_ERROR, DEBUG_CATEGORY_DEBUG, "actual", actual, WIRE_SIZE); + CHECK(0); + } + count++; + } + CHECK(count == test.inserted && sqlite3_step(st) == SQLITE_DONE); + CHECK(sqlite3_finalize(st) == SQLITE_OK); + } +} + +/* Timeout — страховка зависания теста, разрыв связи никогда не ждёт сетевого timeout. */ +static void wait_converged(int callbacks) { + uint64_t deadline = get_time_tb() + 120 * SECOND_TB; + while (test.callbacks < callbacks || !roots_equal()) { + CHECK(get_time_tb() < deadline); + uasync_poll(test.ua, 1); + } + CHECK(test.callbacks == callbacks); + verify_databases(); +} + +/* Меняем память эталона и одну SQLite; листья пачки пересчитываются по одному разу. */ +static void mutate(int writer, const int* indices, int size, int inserting) { + struct node* n = &test.nodes[writer]; + sqlite3* db = n->inst->topo_sqlite_db; + uint64_t prefixes[MAX_BATCH]; + int leaves = 0; + CHECK(size >= 1 && size <= MAX_BATCH); + CHECK(sqlite3_exec(db, "BEGIN", NULL, NULL, NULL) == SQLITE_OK); + for (int i = 0; i < size; i++) { + uint8_t* wire = test.expected[indices[i]].wire; + CHECK((read64(wire + 8) == 0) == inserting); + if (inserting) test.inserted++; + write64(wire + 8, ++test.version); + for (int j = 16; j < WIRE_SIZE; j++) wire[j] = (uint8_t)next_random(); + wire[16] = (uint8_t)writer; + CHECK(put_record(n, wire) == 1); + uint64_t prefix = merkle_sync_level_prefix(read64(wire), MT_MAX_LEVEL); + int p = 0; + while (p < leaves && prefixes[p] != prefix) p++; + if (p == leaves) prefixes[leaves++] = prefix; + } + for (int i = 0; i < leaves; i++) + CHECK(merkle_sync_recompute_path(n->inst, MT_DATA_MEMBERS, TEST_NS, prefixes[i]) == 1); + CHECK(sqlite3_exec(db, "COMMIT", NULL, NULL, NULL) == SQLITE_OK); + merkle_sync_changed(n->inst, MT_DATA_MEMBERS, TEST_NS); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stress: mutate phase=%d batch=%d writer=%d size=%d leaves=%d version=%" PRIu64, + test.phase, test.batch, writer, size, leaves, test.version); +} + +/* После каждого разрыва меняем случайную пачку, затем ждём новый обмен и сверяем обе базы. */ +static void reconnect_phase(int inserting) { + int order[RECORDS]; + shuffled_order(order); + test.phase = inserting ? 1 : 2; + test.batch = 0; + for (int done = 0; done < RECORDS;) { + disconnect_pair(0, 1); + int size = 1 + (int)(next_random() % MAX_BATCH); + if (size > RECORDS - done) size = RECORDS - done; + test.batch++; + mutate(test.batch % 2, order + done, size, inserting); + connect_pair(0, 1); + int callbacks = test.callbacks + 1; + start_pair(0, 1); + wait_converged(callbacks); + done += size; + if (test.batch == 1 || test.batch % 20 == 0 || done == RECORDS) + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: reconnect verified phase=%d batch=%d size=%d done=%d/%d records=%d" + " closes=%d close_max=%.1fms elapsed=%.1fs", test.phase, test.batch, size, done, RECORDS, test.inserted, + test.reconnects, test.disconnect_max / 10.0, (get_time_tb() - test.started) / (double)SECOND_TB); + } +} + +/* Короткий таймер перемежает локальные записи с обработкой настоящих сетевых страниц. */ +static void race_write(void* arg) { + (void)arg; + test.producer = NULL; + if (!roots_equal()) test.race_overlap++; + uint64_t pages = 0; + for (int i = 0; i < 3; i++) pages += test.nodes[i].pages; + if (test.race_left < RACE_BURST && pages > test.last_pages) test.page_interleaves++; + test.last_pages = pages; + int size = 1 + (int)(next_random() % MAX_BATCH), indices[MAX_BATCH]; + for (int i = 0; i < size; i++) { + /* Общий небольшой набор горячих ключей создаёт повторные изменения и старые версии. */ + indices[i] = (int)(next_random() % (i % 4 == 0 ? 32 : RECORDS)); + } + test.batch++; + mutate(test.race_writes % 3, indices, size, 0); + test.race_writes++; + if (--test.race_left) { + test.producer = uasync_set_timeout(test.ua, 10 + next_random() % 40, NULL, race_write, "merkle stress writer"); + CHECK(test.producer != NULL); + } +} + +/* Треугольник остаётся подключённым; после подписки дальнейшие обмены запускает только changed. */ +static void race_phase(void) { + test.phase = 3; + test.batch = 0; + test.count = 3; + create_node(2); + connect_pair(0, 2); + connect_pair(1, 2); + int callbacks = test.callbacks + 2; + start_pair(0, 2); + start_pair(1, 2); + wait_converged(callbacks); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: triangle ready, all three databases contain %d records", test.inserted); + for (int burst = 0; burst < RACE_BATCHES / RACE_BURST; burst++) { + test.race_left = RACE_BURST; + test.producer = uasync_set_timeout(test.ua, 1, NULL, race_write, "merkle stress writer"); + CHECK(test.producer != NULL); + uint64_t deadline = get_time_tb() + 120 * SECOND_TB; + while (test.race_left) { + CHECK(get_time_tb() < deadline); + uasync_poll(test.ua, 1); + } + wait_converged(test.callbacks); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: triangle verified batches=%d/%d mutations=%" PRIu64 + " overlapping=%d page_interleaves=%d elapsed=%.1fs", test.race_writes, RACE_BATCHES, test.version, + test.race_overlap, test.page_interleaves, (get_time_tb() - test.started) / (double)SECOND_TB); + } + CHECK(test.race_overlap > RACE_BATCHES / 2 && test.page_interleaves > RACE_BATCHES / 4); + uint64_t continued = 0, stale = 0; + for (int i = 0; i < 3; i++) { continued += test.nodes[i].continued; stale += test.nodes[i].stale; } + CHECK(continued > 0 && stale > 0); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: race coverage continued_pages=%" PRIu64 " stale_or_duplicate_records=%" PRIu64, + continued, stale); +} + +int main(void) { + debug_config_init(); + debug_set_level(DEBUG_LEVEL_WARN); + /* Полная сверка 10k строк между poll намеренно занимает >10ms; ошибки SYS сохраняем. */ + debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR); + debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO); + if (getenv("UTUN_TEST_DEBUG")) { + debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG); + debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG); + } + test.rng = SEED; + test.started = get_time_tb(); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: start seed=%016" PRIx64 " records=%d batch=1..100 race_batches=%d", + SEED, RECORDS, RACE_BATCHES); + /* 1024 записей одного листа требуют нескольких страниц; остальные распределены по дереву. */ + for (uint32_t i = 0; i < RECORDS; i++) { + uint64_t key = i < 1024 ? i : ((uint64_t)(i * UINT32_C(2654435761)) << 32) | i; + write64(test.expected[i].wire, key); + } + qsort(test.expected, RECORDS, sizeof(test.expected[0]), compare_records); + for (int i = 1; i < RECORDS; i++) CHECK(memcmp(test.expected[i - 1].wire, test.expected[i].wire, 8) < 0); + test.ua = uasync_create(); + CHECK(test.ua != NULL); + test.count = 2; + create_node(0); + create_node(1); + connect_pair(0, 1); + reconnect_phase(1); + reconnect_phase(0); + race_phase(); + CHECK(test.producer == NULL); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: cleanup nodes=%d reconnects=%d callbacks=%d", test.count, test.reconnects, test.callbacks); + /* Освобождаем владельцев без нового poll: destroy ниже сразу закрывает весь транспорт. */ + for (int a = 0; a < test.count; a++) for (int b = 0; b < test.count; b++) { + if (test.links[a][b]) node_conn_direct_close(test.links[a][b]); + test.links[a][b] = NULL; + } + for (int i = 0; i < test.count; i++) { + struct node* n = &test.nodes[i]; + merkle_sync_destroy(n->inst); + CHECK(sqlite3_finalize(n->scan) == SQLITE_OK && sqlite3_finalize(n->write) == SQLITE_OK); + utun_instance_destroy(n->inst); + n->inst = NULL; + } + uasync_poll(test.ua, 0); + CHECK(test.ua->timer_alloc_count == test.ua->timer_free_count); + CHECK(test.ua->socket_alloc_count == test.ua->socket_free_count); + uasync_destroy(test.ua, 0); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "stress: PASS 10000 inserts + 10000 updates with reconnect; 1000 triangle race batches, elapsed=%.1fs", + (get_time_tb() - test.started) / (double)SECOND_TB); + return EXIT_SUCCESS; +}