diff --git a/tests/test_merkle_sync.c b/tests/test_merkle_sync.c index c22ce9f3..b857e841 100644 --- a/tests/test_merkle_sync.c +++ b/tests/test_merkle_sync.c @@ -969,6 +969,9 @@ struct str_state { struct UTUN_INSTANCE* inst[STRESS_N]; struct ms_data data[STRESS_N]; struct ms_test_ctx ctx[STRESS_N]; + uint32_t versions[STRESS_N]; /* Последняя версия каждого владельца записи. */ + unsigned int observer_starts[2]; + uint64_t spam_started_us; void* global_timeout; /* setup: links-up timeout */ void* safety_timer; /* spam+converge safety */ void* stop_timer; /* spam stop */ @@ -996,9 +999,14 @@ static void _str_verify(void); static void _str_fail_timeout(void* arg); static void _str_sync_done(uint64_t peer, const char* ns, int result, void* arg) { - (void)peer; (void)ns; (void)result; struct str_state* s = (struct str_state*)arg; - if (s) __sync_fetch_and_sub(&s->sync_pending, 1); + if (!s || s->phase != 0) return; + if (result != MT_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress initial sync: peer=%016llx ns=%s result=%d", + (unsigned long long)peer, ns, result); + s->phase = 2; + } + __sync_fetch_and_sub(&s->sync_pending, 1); } static void _str_spam_cb(void* arg); @@ -1016,7 +1024,7 @@ static void _str_spam_cb(void* arg) { /* spammer: bump own member, sync to hub */ uint64_t key = ((uint64_t)idx) << 60; - _data_insert(&gs->data[idx], key, (uint32_t)(gs->data[idx].count + 1)); + _data_insert(&gs->data[idx], key, ++gs->versions[idx]); merkle_sync_recompute_path(gs->inst[idx], MT_DATA_MEMBERS, MS_NS_STRESS, key); merkle_sync_start(gs->inst[idx], MT_DATA_MEMBERS, gs->inst[STRESS_HUB]->node_id, MS_NS_STRESS, NULL, NULL); @@ -1027,6 +1035,7 @@ static void _str_obs_spam_cb(void* arg) { if (!gs || !gs->spam_active) return; int idx = (int)(intptr_t)arg; + gs->observer_starts[idx - STRESS_OBS_A]++; merkle_sync_start(gs->inst[idx], MT_DATA_MEMBERS, gs->inst[STRESS_HUB]->node_id, MS_NS_STRESS, NULL, NULL); int delay_tb = ((rand() % 10) + 1) * 10; /* 1-10ms */ @@ -1060,7 +1069,7 @@ static int _str_wait_for(const char* desc, int (*cond)(void), int timeout_tb) { uint64_t start = get_time_tb(); while (!cond() && (get_time_tb() - start) < (uint64_t)timeout_tb && s->phase == 0) uasync_poll(s->ua, 1); - if (cond()) return 1; + if (s->phase == 0 && cond()) return 1; printf(" STRESS TIMEOUT: %s\n", desc); s->phase = 2; return 0; } @@ -1144,14 +1153,15 @@ static int _str_init(void) { /* --- seed each instance with its own member; sync to hub --- */ for (int i = 1; i < STRESS_N; i++) { uint64_t key = ((uint64_t)i) << 60; - _data_insert(&s->data[i], key, (uint32_t)i); + s->versions[i] = (uint32_t)i; + _data_insert(&s->data[i], key, s->versions[i]); merkle_sync_recompute_path(s->inst[i], MT_DATA_MEMBERS, MS_NS_STRESS, key); } /* initial sync: each spoke → hub */ s->sync_pending = STRESS_N - 1; for (int i = 1; i < STRESS_N; i++) merkle_sync_start(s->inst[i], MT_DATA_MEMBERS, s->inst[STRESS_HUB]->node_id, MS_NS_STRESS, _str_sync_done, s); - _str_wait_for("initial sync", _str_cond_sync_done, STRESS_SYNC_TB); + if (!_str_wait_for("initial sync", _str_cond_sync_done, STRESS_SYNC_TB)) goto fail; /* observers: sync with hub too */ @@ -1192,7 +1202,9 @@ static void _str_stop_spam(void* arg) { if (!gs) return; gs->stop_timer = NULL; gs->spam_active = 0; - printf(" spam stopped\n"); + printf(" spam stopped after %lluus; writer updates:", (unsigned long long)(get_time_us() - gs->spam_started_us)); + for (int i = STRESS_SPAM_BEGIN; i <= STRESS_SPAM_END; i++) printf(" %u", gs->versions[i] - (uint32_t)i); + printf("; observer starts: %u %u\n", gs->observer_starts[0], gs->observer_starts[1]); /* короткий дренаж — даём in-flight спам-синкам утихнуть перед финальными раундами */ gs->drain_timer = uasync_set_timeout(gs->ua, STRESS_DRAIN_TB, NULL, _str_drain_cb, "str_drain"); } @@ -1215,8 +1227,14 @@ static void _str_start_final_round(void) { } static void _str_final_done(uint64_t peer, const char* ns, int result, void* arg) { - (void)peer; (void)ns; (void)result; (void)arg; - if (!gs) return; + (void)arg; + if (!gs || gs->phase != 0) return; + if (result != MT_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress final sync: round=%d peer=%016llx ns=%s result=%d", + gs->final_round, (unsigned long long)peer, ns, result); + gs->phase = 2; + return; + } gs->final_pending--; if (gs->final_pending <= 0) _str_check_roots(); } @@ -1225,6 +1243,37 @@ static void _str_verify(void) { struct str_state* s = gs; int ok = 1; + // Нагрузка должна менять записи; равные корни сами по себе не доказывают доставку последних версий. + for (int owner = STRESS_SPAM_BEGIN; owner <= STRESS_SPAM_END; owner++) { + if (s->versions[owner] - (uint32_t)owner >= 10) continue; + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress writer=%d made fewer than ten updates", owner); + ok = 0; + } + if (s->observer_starts[0] < 10 || s->observer_starts[1] < 10) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress observer starts too few: %u %u", + s->observer_starts[0], s->observer_starts[1]); + ok = 0; + } + for (int i = 0; i < STRESS_N && ok; i++) { + if (s->data[i].count != STRESS_N - 1) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress inst=%d items=%d expected=%d", i, s->data[i].count, STRESS_N - 1); + ok = 0; + break; + } + for (int owner = 1; owner < STRESS_N && ok; owner++) { + uint64_t key = ((uint64_t)owner) << 60; + int found = 0; + for (int j = 0; j < s->data[i].count; j++) { + if (s->data[i].items[j].key == key && s->data[i].items[j].val == s->versions[owner]) { found = 1; break; } + } + if (!found) { + DEBUG_ERROR(DEBUG_CATEGORY_MEMBER_SYNC, "stress inst=%d missing final version: owner=%d version=%u", + i, owner, s->versions[owner]); + ok = 0; + } + } + } + /* tree sizes match pairwise */ int na = _db_count_rows(s->inst[STRESS_HUB]->topo_sqlite_db, MS_NS_STRESS); for (int i = 1; i < STRESS_N && ok; i++) { @@ -1314,6 +1363,7 @@ static void test_stress_spam(void) { /* --- start spam --- */ s->spam_active = 1; + s->spam_started_us = get_time_us(); srand(12345); for (int i = STRESS_SPAM_BEGIN; i <= STRESS_SPAM_END; i++) { /* stagger initial delays */