diff --git a/tests/Makefile.am b/tests/Makefile.am index 88a2e431..bb8a2759 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -60,6 +60,7 @@ check_PROGRAMS = \ test_multicast \ test_group_ownership \ test_merkle_protocol \ + test_chat_sync_schedule \ test_ncd_config \ test_node_snapshot \ test_topo_node_wire \ @@ -779,6 +780,10 @@ test_chat_feedback_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) 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_chat_sync_schedule_SOURCES = test_chat_sync_schedule.c $(test_clock_sources) +test_chat_sync_schedule_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/lib +test_chat_sync_schedule_LDFLAGS = $(test_clock_ldflags) +test_chat_sync_schedule_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) diff --git a/tests/test_chat_sync_schedule.c b/tests/test_chat_sync_schedule.c new file mode 100644 index 00000000..6e263fb2 --- /dev/null +++ b/tests/test_chat_sync_schedule.c @@ -0,0 +1,201 @@ +/* Реальные callbacks, БД, lifecycle и uasync; наблюдаем только границу запуска member_sync. + * Private-реализация включена как в test_merkle_protocol: публичный API не расширяем ради теста. */ +#include "chat/member_sync.h" +#include "transport_layer/etcp_api.h" +#include "../lib/u_async.h" +#include "../lib/ll_queue.h" +#include "test_clock.h" +#include +#include +#include + +static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg); +static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry); +static void* schedule_timeout(struct UASYNC* ua, uint64_t tb, void* arg, void (*cb)(void*), const char* name); +#define member_sync_start observe_start +#define etcp_send discard_send +#define uasync_set_timeout schedule_timeout +#include "../src/chat/chat_sync.c" +#undef member_sync_start +#undef etcp_send +#undef uasync_set_timeout + +struct fixture { + struct UASYNC* ua; + struct UTUN_INSTANCE* inst; + struct ETCP_CONN* peers[3]; + char channel[2][21]; + uint64_t gid[2]; +}; + +static struct { uint64_t peer, channel; } starts[128]; +static unsigned start_count; +static int fail_timer; +static struct fixture* injecting; + +static int observe_start(struct UTUN_INSTANCE* inst, uint64_t peer, const char* channel, merkle_sync_done_cb cb, void* arg) { + (void)inst; (void)cb; (void)arg; + assert(start_count < sizeof(starts) / sizeof(starts[0])); + starts[start_count].peer = peer; starts[start_count++].channel = strtoull(channel, NULL, 10); + if (injecting) { + struct fixture* f = injecting; injecting = NULL; + cs_on_conn_down(f->peers[1], ETCP_CBK_EVENT_DOWN, f->inst->chat_sync); + cs_schedule_peer_sync(f->inst->chat_sync, f->peers[2]->peer_node_id, "during_flush"); + } + return 0; +} + +static int discard_send(struct ETCP_CONN* conn, struct ll_entry* entry) { + (void)conn; + queue_dgram_free(entry); queue_entry_free(entry); + return 0; +} + +static void* schedule_timeout(struct UASYNC* ua, uint64_t tb, void* arg, void (*cb)(void*), const char* name) { + if (fail_timer && !strcmp(name, "cs_sync")) { fail_timer = 0; return NULL; } + return uasync_set_timeout(ua, tb, arg, cb, name); +} + +static void run_ms(struct fixture* f, unsigned ms) { test_clock_run_for(f->ua, ms * 10); } + +/* Два канала: A состоит только в первом, B в обоих, C только во втором. */ +static void setup(struct fixture* f) { + memset(f, 0, sizeof(*f)); start_count = 0; fail_timer = 0; injecting = NULL; + f->ua = uasync_create(); assert(f->ua); + f->inst = utun_instance_create_from_str(f->ua, + "[global]\nmy_node_name=schedule test\n[server: udp]\naddr=127.0.0.1:0\ntype=public\n" + "[ntp]\nenabled=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n"); + assert(f->inst && utun_core_start(f->inst) == 0 && chat_service_start(f->inst) == 0); + chat_core_create_channel_auto(f->inst, "schedule 1"); chat_core_create_channel_auto(f->inst, "schedule 2"); + unsigned channels = 0; + for (struct ll_entry* e = f->inst->topo_groups->group_list->head; e; e = e->next) { + struct TOPO_GROUP* group = (struct TOPO_GROUP*)e; + assert(group->group_type == TOPO_GROUP_TYPE_CHAT && channels < 2); + snprintf(f->channel[channels], sizeof(f->channel[channels]), "%s", group->channel_id); + f->gid[channels++] = group->group_id; + } + assert(channels == 2); + for (unsigned i = 0; i < 3; i++) { + struct SC_MYKEYS keys; assert(sc_generate_keypair(&keys) == SC_OK); + struct ETCP_CONN* conn = etcp_connection_create(f->inst, "schedule peer"); assert(conn); + conn->peer_node_id = sc_derive_node_id_from_pubkey(keys.public_key); + conn->initialized = 1; conn->links_up = 1; conn->state = 1; + assert(sc_set_peer_public_key(&conn->crypto_ctx, keys.public_key, SC_PEER_PUBKEY_BIN) == SC_OK); + assert(etcp_conn_index_peer(conn) == 0); f->peers[i] = conn; + uint8_t signature[64] = {0}; + for (unsigned ch = 0; ch < 2; ch++) { + if ((i == 0 && ch == 1) || (i == 2 && ch == 0)) continue; + assert(topo_node_sqlite_member_block_put(f->inst->topo_sqlite_db, f->channel[ch], conn->peer_node_id, + signature, 1, signature, 1, keys.public_key, keys.public_key, "{}", NULL, 0, signature) == 0); + } + } + /* Кеш намеренно не обновляем: задания должны видеть актуальное членство в БД. */ +} + +static void cleanup(struct fixture* f) { + utun_instance_destroy(f->inst); uasync_poll(f->ua, 0); + assert(f->ua->timer_alloc_count == f->ua->timer_free_count); + assert(f->ua->socket_alloc_count == f->ua->socket_free_count); + uasync_destroy(f->ua, 0); +} + +static unsigned peer_starts(struct fixture* f, unsigned peer) { + unsigned count = 0; + for (unsigned i = 0; i < start_count; i++) if (starts[i].peer == f->peers[peer]->peer_node_id) count++; + return count; +} + +static void test_reinit_isolation(void) { + struct fixture f; setup(&f); + for (unsigned i = 0; i < 3; i++) etcp_fire_conn_status(f.peers[i], ETCP_CONN_STATUS_UP); + assert(!start_count); run_ms(&f, 1100); + assert(start_count == 4 && peer_starts(&f, 0) == 1 && peer_starts(&f, 1) == 2 && peer_starts(&f, 2) == 1); + start_count = 0; + for (unsigned i = 0; i < 12; i++) { + etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_REINIT); run_ms(&f, 200); + assert(!peer_starts(&f, 1) && !peer_starts(&f, 2)); + } + run_ms(&f, 1100); assert(start_count == 1 && starts[0].channel == f.gid[0]); + run_ms(&f, 4000); assert(start_count == 1); + cleanup(&f); puts("PASS: repeated REINIT is isolated; unchanged peers and idle service start no new rounds"); +} + +static void test_coalesce_and_deadline(void) { + struct fixture f; setup(&f); + etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP); + for (unsigned i = 0; i < 7; i++) { + run_ms(&f, 100); etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_REINIT); + etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP); + } + run_ms(&f, 400); + assert(start_count == 3 && peer_starts(&f, 0) == 1 && peer_starts(&f, 1) == 2 && !peer_starts(&f, 2)); + cleanup(&f); puts("PASS: duplicate pairs coalesce and other peers cannot move the first deadline"); +} + +static void test_sleep_and_cancel(void) { + struct fixture f; setup(&f); + etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP); + f.peers[0]->peer_sleep_phase = 1; utun_fire_peer_sleep_cbk(f.inst, f.peers[0]->peer_node_id, 1); + run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 1) == 2); + start_count = 0; f.peers[0]->peer_sleep_phase = 0; + utun_fire_peer_sleep_cbk(f.inst, f.peers[0]->peer_node_id, 0); + etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); run_ms(&f, 1100); + assert(start_count == 1 && peer_starts(&f, 0) == 1); + start_count = 0; + etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP); + f.peers[0]->links_up = 0; etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_DOWN); + run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 1) == 2); + start_count = 0; + etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP); + char sql[160]; snprintf(sql, sizeof(sql), "UPDATE peers_%s SET deleted=1 WHERE node_id=%lld", f.channel[0], + (long long)f.peers[1]->peer_node_id); + assert(sqlite3_exec(f.inst->topo_sqlite_db, sql, NULL, NULL, NULL) == SQLITE_OK); + run_ms(&f, 1100); assert(start_count == 1 && starts[0].channel == f.gid[1]); + cleanup(&f); puts("PASS: sleep, DOWN and changed membership cancel only the affected requests"); +} + +static void test_join_existing_transport(void) { + struct fixture f; setup(&f); + cs_send_join_ready(f.inst->chat_sync, f.channel[0], f.peers[0]->peer_node_id); + assert(!start_count); run_ms(&f, 1100); + assert(start_count == 1 && starts[0].peer == f.peers[0]->peer_node_id && starts[0].channel == f.gid[0]); + start_count = 0; + cs_handle_join_ready(f.inst->chat_sync, f.peers[1]->peer_node_id, f.channel[0], (uint8_t*)&f.gid[0], 8); + run_ms(&f, 1100); + assert(start_count == 1 && starts[0].peer == f.peers[1]->peer_node_id && starts[0].channel == f.gid[0]); + cleanup(&f); puts("PASS: both JOIN_READY sides start only their pair without a new transport UP"); +} + +static void test_reentrant_cancel(void) { + struct fixture f; setup(&f); + etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP); + injecting = &f; run_ms(&f, 1100); + assert(start_count == 1 && peer_starts(&f, 0) == 1); + run_ms(&f, 1100); assert(start_count == 2 && peer_starts(&f, 2) == 1 && !peer_starts(&f, 1)); + cleanup(&f); puts("PASS: callbacks can cancel remaining pairs and defer newly queued work"); +} + +static void test_lifecycle_and_timer_failure(void) { + struct fixture f; setup(&f); + fail_timer = 1; etcp_fire_conn_status(f.peers[0], ETCP_CONN_STATUS_UP); + assert(!f.inst->chat_sync->sync_pending && !f.inst->chat_sync->sync_timer); + etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP); + chat_sync_on_channel_deleted(f.inst, f.channel[0]); run_ms(&f, 1100); + assert(start_count == 1 && starts[0].channel == f.gid[1]); + start_count = 0; etcp_fire_conn_status(f.peers[1], ETCP_CONN_STATUS_UP); + chat_service_stop(f.inst); run_ms(&f, 1100); assert(!start_count); + assert(chat_service_start(f.inst) == 0); assert(!start_count); + run_ms(&f, 1100); assert(start_count == 4); + cleanup(&f); puts("PASS: timer failure recovers; channel/stop cancellation and startup retain correct scope"); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); + debug_set_category_level(DEBUG_CATEGORY_MEMBER_SYNC, DEBUG_LEVEL_DEBUG); + test_clock_init(1000); + test_reinit_isolation(); test_coalesce_and_deadline(); test_sleep_and_cancel(); + test_join_existing_transport(); test_reentrant_cancel(); test_lifecycle_and_timer_failure(); + test_clock_finish(); + assert(u_get_allocated_count() == u_get_freed_count()); + puts("ALL PASS: targeted chat synchronization scheduling"); return 0; +}